Re: [PR] KAFKA-17853: Fix termination issue in ConsoleConsumer and ConsoleShareConsumer [kafka]

["lianetm (via GitHub)" <gi...@apache.org>]

lianetm commented on code in PR #19886:
URL: https://github.com/apache/kafka/pull/19886#discussion_r2528979525


##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -229,6 +242,11 @@ protected void checkDisconnects(final long currentTimeMs) {
                 asyncConsumerMetrics.recordUnsentRequestsQueueTime(time.milliseconds() - u.enqueueTimeMs());
                 AuthenticationException authenticationException = client.authenticationException(u.node.get());
                 u.handler.onFailure(currentTimeMs, authenticationException);
+            } else if (u.node.isEmpty() && onClose) {
+                log.debug("Removing unsent request {} because the client is closing", u);
+                iter.remove();
+                asyncConsumerMetrics.recordUnsentRequestsQueueTime(time.milliseconds() - u.enqueueTimeMs());
+                u.handler.onFailure(currentTimeMs, Errors.NETWORK_EXCEPTION.exception());

Review Comment:
   with this we're discarding any request we may have on close , if the node is disconnected, right?
   
   Seems sensible if the node is the Coordinator (similar to what the classic does https://github.com/apache/kafka/blob/b58aae288fb794b01fd2fc6a29748eba1d277bd6/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java#L1142 and the root case of the console issue from what you explain)
   
   But still thinking, could we be wrongly dropping other relevant requests on close, like close fetch sessions? (if let's say, the leader is briefly disconnected and comes back). The ClassicConsumer (and Async before this PR) give those the full closeTimeout 
   https://github.com/apache/kafka/blob/b58aae288fb794b01fd2fc6a29748eba1d277bd6/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L129-L134 
   
   wdyt? I could be missing something on how the flow would actually go in that case.
   
   If we think this could be an issue, I wonder if we could address this at the Coordinator level, to avoid any undesired side-effect on other requests. The CoordinatorReqMgr knows when it's `closing`. Could we consider handling a FindCoordinator failures differently if we know we're closing? here https://github.com/apache/kafka/blob/b58aae288fb794b01fd2fc6a29748eba1d277bd6/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java#L123 (not retrying it anymore). Would that work? 



-- 
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: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org