rishi-rana opened a new pull request, #23444: URL: https://github.com/apache/kafka/pull/23444
## Summary Since #22327 (4.3.1, "MINOR: Fixed metrics decompression"), `ClientTelemetryUtils.decompress` throws `TelemetryTooLargeException` when the decompressed telemetry payload exceeds `telemetry.max.bytes`. That exception propagates up through `ClientTelemetryExporterPlugin.exportMetrics` into `ClientMetricsManager.processPushTelemetryRequest`, where it's caught by a generic `catch (Throwable)` that always maps to `Errors.INVALID_RECORD`. Per KIP-714, `INVALID_RECORD` means "the client's metrics serialization is broken, stop pushing" — the Java client reacts by setting its push interval to `Integer.MAX_VALUE`, permanently disabling telemetry for that client instance until the process restarts. `TELEMETRY_TOO_LARGE` is the correct, retryable code for this case — it's exactly what the pre-decompression wire-size check already returns via `Errors.forException` in `validatePushRequest`, a few lines above in the same method. Typical OTLP metrics decompress at 10-25x their compressed size, so any client sending a payload that's well within the compressed wire limit can still trip the decompressed-size check and get silently, permanently cut off. This was observed in production: after upgrading brokers from 4.3.0 to 4.3.1, every Kafka Streams stream-thread consumer's telemetry stopped in every environment simultaneously. ## Fix Add `catch (TelemetryTooLargeException)` before the existing `catch (Throwable)` in `ClientMetricsManager.processPushTelemetryRequest`, mapping to `Errors.TELEMETRY_TOO_LARGE` instead of `Errors.INVALID_RECORD`. This mirrors the `catch (ApiException) -> Errors.forException(exception)` pattern already used for `validatePushRequest` just above it — no new error-handling shape introduced. `TelemetryTooLargeException` already extends `ApiException` and `Errors.TELEMETRY_TOO_LARGE` already exists (introduced with KIP-714), so this is a two-catch-block change with no protocol/config additions. Out of scope (per the reporter's own note, and to keep this change minimal): whether the decompressed-size bound should have its own configuration or a documented multiple of `telemetry.max.bytes`, since a single value can't sensibly bound both a compressed wire size and a 10-25x larger decompressed size. That's a separate, larger discussion; this PR only fixes the error code mismatch, which is the part causing clients to be permanently and silently cut off. ## Test plan - Added `testPushTelemetryPluginTooLargeException` to `ClientMetricsManagerTest`, mirroring the existing `testPushTelemetryPluginException` (generic-exception) test, asserting the response and the client instance's `lastKnownError` are `TELEMETRY_TOO_LARGE`, with the same plugin-error metric accounting as any other export failure. - `./gradlew :server:test --tests "org.apache.kafka.server.ClientMetricsManagerTest"` — all tests pass. - `./gradlew :server:checkstyleMain :server:checkstyleTest :server:spotlessCheck` — clean. ## AI assistance disclosure Per `AI_POLICY`/`CONTRIBUTING.md`: Claude Code (Sonnet 5) was used to investigate the root cause, implement the fix, and write the test in `server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java` and `server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java`. All changes were reviewed by me before submission. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_01K4B9PPesZhXHiWr5Ak1PNz -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
