[
https://issues.apache.org/jira/browse/KAFKA-21088?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Nilesh Kumar updated KAFKA-21088:
---------------------------------
Description:
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:
{code:java}
} 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();
}
{code}
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.
was:
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.
> 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
> Components: clients
> Reporter: Nilesh Kumar
> Assignee: Nilesh Kumar
> Priority: Minor
> Labels: 3.7.0
>
> 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:
> {code:java}
> } 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();
> }
> {code}
> 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)
