nileshkumar3 opened a new pull request, #23452:
URL: https://github.com/apache/kafka/pull/23452
## Summary
KIP-714 requires a client that would exceed
`GetTelemetrySubscriptionsResponse.TelemetryMaxBytes`
to "reduce the size of the metrics payload so its size does not exceed" that
limit. The broker
sends the value in every subscription response
(`ClientMetricsManager#createGetSubscriptionResponse`)
and rejects oversized pushes with `TELEMETRY_TOO_LARGE`
(`ClientMetricsManager#validatePushRequest`), but `ClientTelemetryReporter`
never read it:
`ClientTelemetrySubscription` did not store the value and
`createPushRequest` collected,
serialized, compressed, and sent the payload with no size check.
On `TELEMETRY_TOO_LARGE` the client only reschedules via
`maybeFetchErrorIntervalMs`, so it
rebuilt a similarly sized payload and re-sent it every push interval, and
the broker rejected it
every time. Nothing in the client ever broke out of that loop.
This keeps `telemetryMaxBytes` in the subscription and enforces it before
sending.
## Changes
- `ClientTelemetryUtils#validateTelemetryMaxBytes`: validates the limit from
the subscription
response, mirroring the existing `validateIntervalMs`. A non-positive
value cannot be honored
(no payload would ever fit), so it logs a warning and is not enforced
rather than silently
disabling telemetry against such a broker. Real brokers always send at
least `1`, since
`telemetry.max.bytes` has valid values `[1,...]`.
- `ClientTelemetrySubscription`: stores `telemetryMaxBytes`, populated from
`GetTelemetrySubscriptionsResponseData#telemetryMaxBytes`, and includes it
in `toString`.
- `ClientTelemetryReporter#createPushRequest`: compares the payload against
the limit after
compression, since the broker validates the compressed bytes carried in
the request. When it
does not fit, the push is skipped and a warning logs the compression type,
metric count,
payload size, and the limit, so an operator can tell whether the
subscription needs narrowing.
The skipped push also transitions to `SUBSCRIPTION_NEEDED` and defers the
next attempt by the
push interval, which is what `handleResponse` already does for a
`TELEMETRY_TOO_LARGE` reply. This
matters for two reasons: returning early without touching state would park
the sender in
`PUSH_IN_PROGRESS` waiting for a response to a request that was never sent,
permanently killing
telemetry; and re-fetching the subscription lets the client pick up a
narrowed metric set or a
raised limit. A terminating push has no valid transition back and nothing to
retry, so its state
is left for `close()` to finish.
I chose the "skip the push" option from the Jira over dropping
lower-priority metric groups: there
is no defined priority order for client metrics to drop, and skipping
already breaks the reject
loop. Dropping groups would be a reasonable follow-up if the warning shows
this happening in
practice.
## Testing
`ClientTelemetryUtilsTest`: valid and invalid limits for
`validateTelemetryMaxBytes`.
`ClientTelemetryReporterTest`: a mocked collector emits one real metric so
the payload is never
empty, then covers a payload within the limit still being sent, an oversized
payload being skipped
with the state and interval above and telemetry left enabled, and an
oversized terminating push
being skipped without disabling telemetry. Also asserts the limit survives
the subscription
response, including the invalid-value fallback.
--
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]
