GH-2576: Fix No Seeks After Error Handling #2579
Merged
Conversation
This file contains hidden or 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
Resolves spring-projects#2576 When the error handler `seeksAfterError` is false, we keep the unprocessed records in memory (`retainedRecords`), pause the consumer, and attempt redelivery. If a rebalance occurs during this state, the behavior depends on whether a legacy or cooperative partition assignor is used. With a legacy assignor, all partitions are unassigned and usually a subset is reassigned. This causes their positions to be reset to the last committed offset. In this case, all retained records must be cleared, otherwise an emergency stop occurs because records are polled when we don't expect them. With a cooperative assignor, only a subset of the partitions are revoked. Records for these partitions must be pruned from the `retainedRecords` and not redelivered to the listener. - add code to the rebalance listener to prune the revoked paritions from the retained records - re-pause any assigned partitions if retained records are present Add tests for both types of assignor. Also tested with the reporter's reproducer. **cherry-pick to 2.9.x**
artembilan
requested changes
Feb 14, 2023
...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
Outdated
Show resolved
Hide resolved
...ng-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
Outdated
Show resolved
Hide resolved
Co-authored-by: Artem Bilan <[email protected]>
|
Cherry-picked to |
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.
Resolves #2576
When the error handler
seeksAfterErroris false, we keep the unprocessed records in memory (retainedRecords), pause the consumer, and attempt redelivery.If a rebalance occurs during this state, the behavior depends on whether a legacy or cooperative partition assignor is used.
With a legacy assignor, all partitions are unassigned and usually a subset is reassigned. This causes their positions to be reset to the last committed offset. In this case, all retained records must be cleared, otherwise an emergency stop occurs because records are polled when we don't expect them.
With a cooperative assignor, only a subset of the partitions are revoked. Records for these partitions must be pruned from the
retainedRecordsand not redelivered to the listener.Add tests for both types of assignor.
Also tested with the reporter's reproducer.
cherry-pick to 2.9.x