Nilesh Kumar created KAFKA-21088:
------------------------------------
Summary: Client telemetry stops permanently if push payload
construction fails
Key: KAFKA-21088
URL: https://issues.apache.org/jira/browse/KAFKA-21088
Project: Kafka
Issue Type: Bug
Reporter: Nilesh Kumar
ClientTelemetryReporter.DefaultClientTelemetrySender can be left in the
PUSH_IN_PROGRESS
state with no request in flight, after which no further telemetry request is
ever made for
the lifetime of the client.
createPushRequest(ClientTelemetrySubscription) transitions the sender to
PUSH_IN_PROGRESS
before the payload is assembled. The assembly step then catches any exception,
logs it,
calls updateErrorResult, and returns Optional.empty() without restoring the
state:
} catch (Exception e) {
log.warn("Error constructing client telemetry payload: ", e);
// Update last accessed time for push request to be retried on next
interval.
updateErrorResult(localSubscription.pushIntervalMs,
time.milliseconds());
return Optional.empty();
}
The comment says the push will be retried on the next interval, but there is no
next
attempt:
- No request was sent, so handleResponse(PushTelemetryResponse) is never
invoked and cannot
move the state to PUSH_NEEDED.
- No request failed, so handleFailedPushTelemetryRequest / handleFailedRequest
is never
invoked and cannot move the state to SUBSCRIPTION_NEEDED.
- timeToNextUpdate returns requestTimeoutMs for PUSH_IN_PROGRESS, which is
greater than
zero, so NetworkClient.TelemetrySender.maybeUpdate returns early and never
calls
createRequest() again.
The sender is therefore stuck. The only exit is initiateClose()/close() on
client shutdown,
which is not a recovery. Telemetry silently stops after a single WARN log, while
ClientTelemetrySender remains enabled, so nothing reports that metrics are no
longer being
delivered.
Reachability is low. KafkaMetricsCollector.collect guards each metric with its
own
try/catch and collectMetric separately guards metric.metricValue(), so a
misbehaving metric
does not trigger this. The remaining triggers are a runtime exception from
createPayload
(protobuf assembly) or from the ClientTelemetryEmitter init/close in the
try-with-resources.
Errors such as OutOfMemoryError are not caught here and propagate instead.
For contrast, the compression failure path immediately below is not affected:
it either
falls back to uncompressed data and continues, or sets TERMINATED and throws.
The
kafkaMetricsCollector == null path is also fine, because it returns before the
state is
changed.
Suggested fix
-------------
Restore a state from which a retry can happen before returning, i.e. transition
back to
PUSH_NEEDED (or SUBSCRIPTION_NEEDED, matching how a failed push is handled
elsewhere) when
not terminating, so that the push interval retry the comment describes actually
occurs. For
a terminating push, returning empty is acceptable since close() follows.
The underlying invariant worth asserting in a test: the sender should never
remain in
PUSH_IN_PROGRESS or TERMINATING_PUSH_IN_PROGRESS after createRequest() returns
Optional.empty(), because no response or failure callback can arrive to move it
on.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
