KAFKA-19357: AsyncConsumer#close hangs during closing because the commitAsync request never completes due to a missing coordinator #19914
Conversation
github-actions
bot
added
triage
| @@ -99,7 +99,7 @@ public void signalClose() { | |||
| */ | |||
| @Override | |||
| public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { | |||
| if (closing || this.coordinator != null) | |||
There was a problem hiding this comment.
This makes the consumer find the coordinator during closing, right? If the consumer doesn't have a coordinator running, does it make sense to find one during closing?
There was a problem hiding this comment.
Well, we do need to be careful. There is also code to make sure that finding a coordinator does not block the progress of closing. If you start a consumer when there are no running brokers, close needs to complete promptly. If you stop the brokers when a consumer has been running and then attempt a close, it also needs to complete promptly.
| @@ -99,7 +99,7 @@ public void signalClose() { | |||
| */ | |||
| @Override | |||
| public NetworkClientDelegate.PollResult poll(final long currentTimeMs) { | |||
| if (closing || this.coordinator != null) | |||
| if (this.coordinator != null) | |||
There was a problem hiding this comment.
Thanks for the PR, left a comment.
Should we add an IT for this scenario: when closing the consumer, commitAsync, and verify that the consumer can shut down properly?
Problem:
When AsyncConsumer is closing, CoordinatorRequestManager stops looking for coordinator by returning EMPTY in poll() method when closing flag is true.
This prevents commitAsync() and other coordinator-dependent operations from completing, causing close() to hang indefinitely.
Solution:
Remove the closing flag check in poll() method of CoordinatorRequestManager, so it continues to look for coordinator when needed, even during closing state. This ensures pending coordinator-dependent operations can complete during shutdown.