KAFKA-16599: LegacyConsumer should always await pending async commits on commitSync and close #15693
Conversation
…itSync and close (#15613) The javadoc for KafkaConsumer.commitSync says: Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)} (or similar) are guaranteed to have their callbacks invoked prior to completion of this method. This is not always true in the async consumer, where there is no code at all to make sure that the callback is executed before commitSync returns. Similarly, the async consumer is also missing logic to await callback execution in close. While the javadoc doesn't explicitly promise callback execution, it promises "completing commits", which one would reasonably expect to include callback execution. Also, the legacy consumer contains some code to execute callbacks before closing. This change proposed a number of fixes to clean up the callback execution guarantees in the async consumer: We keep track of the incomplete async commit futures and wait for them to complete before returning from commitSync or close (if there is time). Since we need to block to make sure that our previous commits are completed, we allow the consumer to wake up. Some similar gaps are addressed in the legacy consumer, see #15693 Testing Two new integration tests and a couple of unit tests. Reviewers: Bruno Cadonna <cadonna@apache.org>, Kirk True <ktrue@confluent.io>, Lianet Magrans <lianetmr@gmail.com>
lucasbru
changed the title
lucasbru
force-pushed
the
kip848_await_async_commit_legacy
branch
from
a666908
to
59b2e10
Compare
| // Super-class close may wait for more commit callbacks to complete. | ||
| invokeCompletedOffsetCommitCallbacks(); |
There was a problem hiding this comment.
I'm not familiar enough with this code. What situations would we want to invoke callbacks after we've closed the coordinator?
There was a problem hiding this comment.
There may async commits that are completed during coordinator close. This is done to provide the guarantee that we want to complete async commits in Consumer.close (as long as we don't timeout). We also provide the guarantee that if an async commit completes, the corresponding callback is executed. So then we need to execute callbacks here.
This change could cause issues if we somehow interact with the consumer within the commit callback, but the consumer is already partially closed. Note that callback is already called during close (further above) where the coordinator isn't called yet. So essentially, this change will just make it "more commonplace" to call the callbacks during close.
But it would be an option to just leave the legacy consumer to have somewhat inconsistent behavior here, and just change it for the new consumer. WDYT?
| pendingAsyncCommits.decrementAndGet(); | ||
| completedOffsetCommits.add(new OffsetCommitCompletion(callback, offsets, | ||
| try { | ||
| completedOffsetCommits.add(new OffsetCommitCompletion(callback, offsets, | ||
| new RetriableCommitFailedException(e))); | ||
| } finally { | ||
| pendingAsyncCommits.decrementAndGet(); | ||
| } |
There was a problem hiding this comment.
I understand moving the counter decrement to a finally block, but are we expecting that the add() method or constructor would throw an exception?
There was a problem hiding this comment.
No, this is just defensive programming
| commitException = new RetriableCommitFailedException(e); | ||
| } | ||
| completedOffsetCommits.add(new OffsetCommitCompletion(cb, offsets, commitException)); | ||
| if (commitException instanceof FencedInstanceIdException) { |
There was a problem hiding this comment.
I know you didn't write the original logic, but it seems funny to compare the exception against a FencedInstanceIdException here if up above commitException was converted to a RetriableCommitFailedException. I guess there's not a better way to construct this without potentially changing the logic to be incorrect.
There was a problem hiding this comment.
FencedInstanceIdException is not retriable, so this shouldn't make a difference, right?
The javadoc for
KafkaConsumer.commitSyncsays:This is not always true in the legacy consumer, when the set of offsets is empty, the execution of the commit callback is not always awaited. There are also various races possible that can avoid callback handler execution.
Similarly, there is code in the legacy consumer to await the completion of the commit callback before closing, however, the code doesn't cover all cases and the behavior is therefore inconsistent. While the javadoc doesn't explicitly promise callback execution, it promises "completing commits", which one would reasonably expect to include callback execution. Either way, the current behavior of the legacy consumer is inconsistent.
This change proposed a number of fixes to clean up the callback execution guarantees:
Committer Checklist (excluded from commit message)