KAFKA-19357: AsyncConsumer#close hangs as commitAsync never completes when coordinator is missing

[Mirai1129]

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 until timeout.

Solution:
Modified the closing flag check in poll() method of
CommitRequestManager to be more targeted:

• When both coordinators are unknown and the consumer is closing, only
  return EMPTY
• When this condition is met, proactively fail all pending commit
  requests with CommitFailedException
• This allows coordinator lookup to continue when coordinator is
  available during shutdown, while preventing indefinite hangs when
  coordinator is unreachable

[chia7712]

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?

[AndrewJSchofield]

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]

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.

[Mirai1129]

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?

[frankvicky]

@Mirai1129: Thanks for the update.

[m1a2st]

Thank @Mirai1129 for this patch, left some comments

[TaiJuWu]

LGTM assuming CI pass

[m1a2st]

Why this callback need this comment?

[Mirai1129]

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!

[TaiJuWu]

@Mirai1129 could you merge trunk to get fix testMultipleRemoteFetchesInOneFetchRequest()?

[Yunyung]

How about

Suggested change

	consumer.close(CloseOptions.timeout(Duration.ofMillis(500)));
	assertDoesNotThrow(() -> consumer.close(CloseOptions.timeout(Duration.ofMillis(1000))));

[frankvicky]

It doesn't add much value to this test case.

[Yunyung]

LGTM. Could you update the PR description?

[Mirai1129]

@Yunyung I've updated the solution! Could you please take a look at the changes when you have a moment? Thanks!

[Yunyung]

causing close() to hang indefinitely.

Not very accurate, it should be "causing close() to hang until timeout".

Modified the closing flag check in poll() method of
CoordinatorRequestManager to be more targeted:

CoordinatorRequestManager -> CommitRequestManager

[Mirai1129]

@Yunyung Thank you so much, I have updated the description.