KAFKA-12870; Flush in progress not cleared after transaction completion #10880
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
Note I have a few additional tests which I will post shortly. |
guozhangwang
approved these changes
Jun 18, 2021
| ); | ||
| } | ||
|
|
||
| private void runUntil(Sender sender, Supplier<Boolean> condition) { |
There was a problem hiding this comment.
I remember we have similar utils in TransactionManagerTest, could we consolidate them?
There was a problem hiding this comment.
Would it be reasonable to add a runUntil helper to Sender for testing?
There was a problem hiding this comment.
Yup, I think that's also fine.
There was a problem hiding this comment.
Decided to pull out a test util instead. Hopefully that is ok.
hachikuji
added a commit
that referenced
this pull request
Jun 18, 2021
…on (#10880) We had been using `RecordAccumulator.beginFlush` in order to force the `RecordAccumulator` to flush pending batches when a transaction was being completed. Internally, `RecordAccumulator` has a simple counter for the number of flushes in progress. The count gets incremented in `beginFlush` and it is expected to be decremented by `awaitFlushCompletion`. The second call to decrement the counter never happened in the transactional path, so the counter could get stuck at a positive value, which means that the linger time would effectively be ignored. This patch fixes the problem by removing the use of `beginFlush` in `Sender`. Instead, we now add an additional condition in `RecordAccumulator` to explicitly check when a transaction is being completed. Reviewers: Guozhang Wang <wangguoz@gmail.com>
xdgrulez
pushed a commit
to xdgrulez/kafka
that referenced
this pull request
Dec 22, 2021
…on (apache#10880) We had been using `RecordAccumulator.beginFlush` in order to force the `RecordAccumulator` to flush pending batches when a transaction was being completed. Internally, `RecordAccumulator` has a simple counter for the number of flushes in progress. The count gets incremented in `beginFlush` and it is expected to be decremented by `awaitFlushCompletion`. The second call to decrement the counter never happened in the transactional path, so the counter could get stuck at a positive value, which means that the linger time would effectively be ignored. This patch fixes the problem by removing the use of `beginFlush` in `Sender`. Instead, we now add an additional condition in `RecordAccumulator` to explicitly check when a transaction is being completed. Reviewers: Guozhang Wang <wangguoz@gmail.com>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
We had been using
RecordAccumulator.beginFlushin order to force theRecordAccumulatorto flush pending batches when a transaction was being completed. Internally,RecordAccumulatorhas a simple counter for the number of flushes in progress. The count gets incremented inbeginFlushand it is expected to be decremented byawaitFlushCompletion. The second call to decrement the counter never happened in the transactional path, so the counter could get stuck at a positive value, which means that the linger time would effectively be ignored.The patch here fixes the problem by removing the use of
beginFlushinSender. Instead, we now add an additional condition inRecordAccumulatorto explicitly check when a transaction is being completed.Committer Checklist (excluded from commit message)