[
https://issues.apache.org/jira/browse/KAFKA-21076?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Apoorv Mittal resolved KAFKA-21076.
-----------------------------------
Fix Version/s: 4.5.0
Resolution: Fixed
> Oversize decompressed telemetry payload returns INVALID_RECORD and
> permanently disables client telemetry
> --------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21076
> URL: https://issues.apache.org/jira/browse/KAFKA-21076
> Project: Kafka
> Issue Type: Bug
> Components: clients, core
> Affects Versions: 4.3.1
> Reporter: Steven Schlansker
> Assignee: Rishi Kant Rana
> Priority: Major
> Fix For: 4.5.0
>
>
> Since [PR #22327|https://github.com/apache/kafka/pull/22327] (4.3.1, "MINOR:
> Fixed metrics decompression"), {{ClientTelemetryUtils.decompress}} throws
> {{TelemetryTooLargeException}} when the decompressed payload exceeds
> {{{}telemetry.max.bytes{}}}. The call runs inside
> {{ClientMetricsManager.processPushTelemetryRequest}} in the plugin export
> block, whose {{catch (Throwable)}} maps every failure to
> {{{}Errors.INVALID_RECORD{}}}:
> {code:java}
> catch (Throwable exception) {
> clientMetricsStats.recordPluginErrorCount(clientInstanceId);
> clientInstance.lastKnownError(Errors.INVALID_RECORD);
> log.error("Error exporting client metrics to the plugin for client
> instance id: {}", clientInstanceId, exception);
> return request.errorResponse(0, Errors.INVALID_RECORD);
> }
> {code}
> KIP-714 defines INVALID_RECORD as "Broker failed to decode or validate the
> client's encoded metrics. Log an error and stop pushing metrics. This is
> viewed as a problem in the client implementation of metrics serialization
> that is not likely to be resolved by retrying." The Java client implements
> exactly that in {{{}ClientTelemetryUtils.maybeFetchErrorIntervalMs{}}}: on
> INVALID_RECORD it sets the push interval to {{{}Integer.MAX_VALUE{}}}, so
> telemetry from that client instance stops until the process restarts.
> The validatePushRequest method throws the same TelemetryTooLargeException
> when the wire size exceeds telemetry.max.bytes, and that path returns the
> error through Errors.forException, which yields TELEMETRY_TOO_LARGE. So the
> broker returns two different codes for the same exception class depending on
> whether the check happens before or after decompression.
> KIP-714 defines TELEMETRY_TOO_LARGE for a payload that is too large, and the
> client handles that code by retrying at the normal interval.
> Observed effect: after a broker upgrade from 4.3.0 to 4.3.1, every Kafka
> Streams stream-thread consumer (payload several MB decompressed, under 1 MiB
> compressed) got INVALID_RECORD on its next push and disabled its telemetry.
> All {{org.apache.kafka.stream.*}} metrics that KIP-1076 routes through the
> stream-thread consumers disappeared from the metrics backend in every
> environment at the same time. Broker log:
> {noformat}
> ERROR Error exporting client metrics to the plugin for client instance id:
> xkdFrlXGTGig-RPx5SjuQA (org.apache.kafka.server.ClientMetricsManager)
> org.apache.kafka.common.errors.TelemetryTooLargeException: Decompressed
> telemetry metrics exceed maximum allowed size: 1048576
> {noformat}
> Two further points:
> * KIP-714 documents {{telemetry.max.bytes}} as "The maximum size (after
> compression if compression is used) of telemetry pushed from a client to the
> broker." PR #22327 changed the meaning to also bound the decompressed size,
> with no KIP update, no release note, and no new configuration. A 4.3.0
> cluster that accepted a client's pushes rejects the same pushes on 4.3.1.
> * Before PR #22327 the same code path allocated {{metrics.limit() * 2}}
> bytes and grew without bound, so a bound is reasonable. The bound needs its
> own error code and, given that decompression ratios of 10x or more are normal
> for OTLP metrics, its own configuration or a multiple of
> {{{}telemetry.max.bytes{}}}.
> Measured on a Kafka Streams application (8 stream threads, subscription
> metrics=*, push interval 30 s) against 4.3.1 brokers. The broker's
> RequestMetrics RequestBytes histogram for PushTelemetry put the compressed
> push at a 40 KB mean, 940 KB p95, 1.34 MB p99 and 1.41 MB max in one cluster,
> 2.08 MB max in another; all of these passed the 1 MiB wire check before 4.3.1
> or were rejected with the retryable TELEMETRY_TOO_LARGE. With
> telemetry.max.bytes raised to 32 MiB the brokers still logged "Decompressed
> telemetry metrics exceed maximum allowed size: 33554432" for several of these
> pushes, and 128 MiB was needed before every push was accepted. That is an
> expansion ratio of at least 25x between the compressed payload the client and
> the wire check see and the decompressed payload the new check bounds, which
> is the normal ratio for OTLP metrics with repeated label sets. A single
> telemetry.max.bytes cannot bound both sensibly: a value that admits a
> legitimate decompressed payload is meaningless as a wire limit. The
> decompressed bound should have its own configuration, or a documented
> multiple of telemetry.max.bytes, and it should return TELEMETRY_TOO_LARGE
> like the wire check does.
> Suggested fix: catch {{TelemetryTooLargeException}} before the generic
> {{catch (Throwable)}} and return {{{}Errors.TELEMETRY_TOO_LARGE{}}}, so the
> client retries at its interval instead of disabling telemetry. Document the
> decompressed bound in {{telemetry.max.bytes}} or add a separate
> configuration, and mention the behavior change in the upgrade notes.
> Workaround: raise {{telemetry.max.bytes}} on the brokers and restart affected
> clients.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
