[jira] [Commented] (KAFKA-21076) Oversize decompressed telemetry payload returns INVALID_RECORD and permanently disables client telemetry

["Rishi Kant Rana (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-21076?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18114916#comment-18114916 ] 

Rishi Kant Rana commented on KAFKA-21076:
-----------------------------------------

PR Raised [https://github.com/apache/kafka/pull/23444/] 

> 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
>
> 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)