[
https://issues.apache.org/jira/browse/KAFKA-21076?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Steven Schlansker updated KAFKA-21076:
--------------------------------------
Description:
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.
was:
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.
> 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
> 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)
