Fix batch listener backoff state reset in multi-partition containers - #4518
Open
29shivam wants to merge 1 commit into
Open
Fix batch listener backoff state reset in multi-partition containers#451829shivam wants to merge 1 commit into
29shivam wants to merge 1 commit into
Conversation
In a non-concurrent container with multiple partitions, a successful batch invocation after a prior failure called clearThreadState() on the CommonErrorHandler, wiping FailedRecordTracker state for ALL partitions. This caused partitions that were still inside their backoff/retry cycle to lose their failure count and restart from attempt 1, preventing broken records from ever reaching the DLQ. Root cause: clearThreadState() removes the entire Thread->Map entry from FailedRecordTracker, regardless of which partitions were actually part of the successful batch. Fix: introduce clearThreadStateFor(Collection<TopicPartition>) so the container can scope the reset to only the partitions present in the current successful batch. doInvokeBatchListener() now calls this narrower method, leaving backoff state intact for partitions that are still retrying. The full clearThreadState() path (called on thread termination) is unchanged. Composite handlers (CommonMixedErrorHandler, CommonDelegatingErrorHandler) delegate the new method to their children. Closes spring-projectsgh-4371 Signed-off-by: 29shivam <shivamsingh6126@gmail.com>
29shivam
force-pushed
the
fix/batch-listener-backoff-partition-scoped-clear
branch
from
June 26, 2026 04:20
174b0df to
1470bba
Compare
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
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.
Problem
In a non-concurrent
KafkaMessageListenerContainerwith multiple partitions, when a successful batch follows a prior failure,doInvokeBatchListenercallscommonErrorHandler.clearThreadState(). This wipes the entireFailedRecordTrackermap for the current thread, resetting the failure count for all partitions — including partitions that are still inside their backoff/retry cycle.The result: a broken record that should be retried N times (or sent to the DLQ after N attempts) restarts its counter every time any other partition successfully processes a batch, so it never exhausts its retries and never reaches the DLQ.
Reproduction scenario (from #4371):
batchFailed = true,FailedRecordTrackerrecords attempt 1 for AclearThreadState()→ A's failure count wipedFix
Introduce
clearThreadStateFor(Collection<TopicPartition> partitions)throughout the error-handler stack:FailedRecordTracker— removes only the specified partitions from the per-thread failure mapFailedRecordProcessor— public delegate to trackerCommonErrorHandler— new default (no-op) so existing custom handlers need no changesCommonMixedErrorHandler/CommonDelegatingErrorHandler— delegate to sub-handlersKafkaMessageListenerContainer.doInvokeBatchListener— callsclearThreadStateFor(records.partitions())instead ofclearThreadState()This scopes the reset to only the partitions that were actually part of the successful batch, leaving backoff state intact for partitions still retrying.
The existing
clearThreadState()call (on thread termination) is unchanged.Checklist
@sincetag added to new API (CommonErrorHandler.clearThreadStateFor)Closes #4371