[jira] [Commented] (KAFKA-21088) Client telemetry stops permanently if push payload construction fails

["Apoorv Mittal (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-21088?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18115095#comment-18115095 ] 

Apoorv Mittal commented on KAFKA-21088:
---------------------------------------

[~nileshk03] Thanks for raising the issue. I can see that you assigned the ticket yourself so in case you are not working on the fix then feel free to assign it to me. Else I can review the changes.

> 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.
> h3. 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)