KAFKA-19357: AsyncConsumer#close hangs during closing because the commitAsync request never completes due to a missing coordinator #19914

Mirai1129 · 2025-06-06T14:54:09Z

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.

chia7712 · 2025-06-06T15:03:54Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java

@@ -99,7 +99,7 @@ public void signalClose() {
     */
    @Override
    public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
-        if (closing || this.coordinator != null)


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?

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.

DL1231 · 2025-06-06T15:33:49Z

...nts/src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java

@@ -99,7 +99,7 @@ public void signalClose() {
     */
    @Override
    public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
-        if (closing || this.coordinator != null)
+        if (this.coordinator != null)


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?

kirktrue

Thanks for the PR @Mirai1129! As @DL1231 mentioned, because closing the AsyncKafkaConsumer is such a tricky area, we really do need to have some tests to assure us that this works and doesn't break something else.

Merge branch 'trunk' into KAFKA-19357

…inatorRequestManager`

Mirai1129 · 2025-07-03T13:13:58Z

Hi everyone,

Thanks for all the feedback and suggestions! And sorry for late reply. I've updated the PR with the timeout handling solution - when CommitRequestManager times out, it will now immediately clear remaining callbacks to prevent system deadlock.

I'm currently working on adding comprehensive integration tests to cover this scenario and will push them shortly.

frankvicky

Given that the coordinator might keep an unknown state and probably won't recover in time.
To prevent blocking the closing process, we should complete the pending commits request exceptionally (e.g., CommitFailedException) if the consumer is closing and the coordinator is unknown.
This could inform the user that the last commits have not been submitted correctly and also ensure a smooth closing process.
Thoughts?

…is unknown

frankvicky

@Mirai1129: Thanks for the update.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

m1a2st

Thank @Mirai1129 for this patch, left some comments

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

TaiJuWu

LGTM assuming CI pass

m1a2st · 2025-07-14T13:54:52Z

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

+            // Try without looking up the coordinator first
+            var callback = new CountConsumerCommitCallback();


Why this callback need this comment?

Actually idk why, cuz I just duplicated testCommitAsyncCompletedBeforeConsumerCloses and modified it. But if it doesn't need let me remove it 🥹.

kirktrue

Looks good to me. Just a couple of questions on the test. Thanks!

kirktrue · 2025-07-15T00:22:29Z

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

+            var callback = new CountConsumerCommitCallback();
+
+            // Close the coordinator before committing because otherwise the commit will fail to find the coordinator.
+            cluster.brokerIds().forEach(cluster::shutdownBroker);


I wonder if we need to have some sort of wiatFor() check after this to ensure the brokers are down.

Thank you so much! I'll add waitFor() ensure the brokers are down.

kirktrue · 2025-07-15T00:24:09Z

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java

+
+            consumer.poll(Duration.ofMillis(500));
+            consumer.commitAsync(Map.of(tp1, new OffsetAndMetadata(1L)), callback);
+            consumer.close(CloseOptions.timeout(Duration.ofMillis(500)));


Is it possible for close() to take more than 500 milliseconds? Should we time the method call and add an assert to ensure it's not more than, say, 1000 milliseconds?

Is this change to avoid the test being incorrect due to the shutdown taking too long then we add an assert to ensure that?

fix: Fix AsyncConsumer close hanging when coordinator is needed

bb64968

github-actions bot added triage PRs from the community consumer clients small Small PRs labels Jun 6, 2025

chia7712 reviewed Jun 6, 2025

View reviewed changes

DL1231 reviewed Jun 6, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Jun 7, 2025

Merge branch 'trunk' into KAFKA-19357

a7d1d45

kirktrue suggested changes Jul 1, 2025

View reviewed changes

Mirai1129 added 2 commits July 3, 2025 08:03

Merge branch 'trunk' into KAFKA-19357

2985c97

Merge branch 'trunk' into KAFKA-19357

clear request when coordinator is empty and is closing, revert `Coord…

24a0bbd

…inatorRequestManager`

frankvicky reviewed Jul 11, 2025

View reviewed changes

frankvicky added the ci-approved label Jul 11, 2025

Mirai1129 added 2 commits July 13, 2025 06:45

Merge branch 'trunk' into KAFKA-19357

26cbc2d

fix: Handle commit failures when consumer is closing and coordinator …

5a5511f

…is unknown

frankvicky reviewed Jul 13, 2025

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java Outdated Show resolved Hide resolved

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java Outdated Show resolved Hide resolved

m1a2st reviewed Jul 13, 2025

View reviewed changes

frankvicky reviewed Jul 13, 2025

View reviewed changes

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java Outdated Show resolved Hide resolved

Mirai1129 added 6 commits July 13, 2025 07:37

fix: check exception message and inline var

da08d36

fix: inline var

5fef9bf

fix: uncomment other line

ba9eb68

fix: close by CloseOptions not Duration

5c72507

fix: remove duplicated operation and docs position

7d7aa25

fix: change assertEquals to exception count not success count

da28cae

TaiJuWu reviewed Jul 14, 2025

View reviewed changes

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java Outdated Show resolved Hide resolved

fix: remove one partition to check coordinator is not found

8145ba2

Mirai1129 requested a review from chia7712 July 14, 2025 09:24

Mirai1129 requested review from TaiJuWu, frankvicky and m1a2st July 14, 2025 09:24

TaiJuWu reviewed Jul 14, 2025

View reviewed changes

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java Outdated Show resolved Hide resolved

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java Show resolved Hide resolved

frankvicky reviewed Jul 14, 2025

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java Outdated Show resolved Hide resolved

...ation-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java Show resolved Hide resolved

fix: remove multiple commitAsync and redundant comment

03139eb

TaiJuWu approved these changes Jul 14, 2025

View reviewed changes

m1a2st reviewed Jul 14, 2025

View reviewed changes

Mirai1129 added 2 commits July 14, 2025 14:00

Merge branch 'trunk' into KAFKA-19357

1155e7e

fix: remove unneeded comment

44cf95d

kirktrue reviewed Jul 15, 2025

View reviewed changes

		// Try without looking up the coordinator first
		var callback = new CountConsumerCommitCallback();

KAFKA-19357: AsyncConsumer#close hangs during closing because the commitAsync request never completes due to a missing coordinator #19914

Are you sure you want to change the base?

KAFKA-19357: AsyncConsumer#close hangs during closing because the commitAsync request never completes due to a missing coordinator #19914

Conversation

Mirai1129 commented Jun 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Mirai1129 commented Jul 3, 2025

Uh oh!

frankvicky left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankvicky left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

m1a2st left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

TaiJuWu left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Mirai1129 Jul 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Mirai1129 commented Jun 6, 2025 •

edited

Loading

frankvicky left a comment •

edited

Loading

Mirai1129 Jul 14, 2025 •

edited

Loading