From 1470bba7f4539dfa4a90d0a023befedce680639a Mon Sep 17 00:00:00 2001 From: 29shivam Date: Thu, 25 Jun 2026 23:16:59 -0500 Subject: [PATCH] Fix batch listener backoff state reset in multi-partition containers 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) 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 gh-4371 Signed-off-by: 29shivam --- .../kafka/listener/CommonDelegatingErrorHandler.java | 9 +++++++++ .../kafka/listener/CommonErrorHandler.java | 9 +++++++++ .../kafka/listener/CommonMixedErrorHandler.java | 8 ++++++++ .../kafka/listener/FailedRecordProcessor.java | 6 ++++++ .../kafka/listener/FailedRecordTracker.java | 7 +++++++ .../kafka/listener/KafkaMessageListenerContainer.java | 2 +- 6 files changed, 40 insertions(+), 1 deletion(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java index 5febeb4b13..8e69d24f31 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java @@ -16,12 +16,15 @@ package org.springframework.kafka.listener; +import java.util.Collection; import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import org.apache.kafka.common.TopicPartition; + import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; @@ -97,6 +100,12 @@ public void clearThreadState() { this.delegates.values().forEach(CommonErrorHandler::clearThreadState); } + @Override + public void clearThreadStateFor(Collection partitions) { + this.defaultErrorHandler.clearThreadStateFor(partitions); + this.delegates.values().forEach(h -> h.clearThreadStateFor(partitions)); + } + @Override public boolean isAckAfterHandle() { return this.defaultErrorHandler.isAckAfterHandle(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java index 4291ca6fa6..c50d28fe24 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonErrorHandler.java @@ -165,6 +165,15 @@ default int deliveryAttempt(TopicPartitionOffset topicPartitionOffset) { default void clearThreadState() { } + /** + * Clear thread state only for the given partitions. Called after a successful batch + * invocation to avoid resetting backoff state for partitions that are still retrying. + * @param partitions the partitions whose retry state should be cleared. + * @since 4.0 + */ + default void clearThreadStateFor(Collection partitions) { + } + /** * Return true if the offset should be committed for a handled error (no exception * thrown). diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java index d9c7cb7ff1..7d5b143222 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonMixedErrorHandler.java @@ -16,11 +16,13 @@ package org.springframework.kafka.listener; +import java.util.Collection; import java.util.List; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.common.TopicPartition; import org.springframework.kafka.support.TopicPartitionOffset; import org.springframework.util.Assert; @@ -105,6 +107,12 @@ public void clearThreadState() { this.recordErrorHandler.clearThreadState(); } + @Override + public void clearThreadStateFor(Collection partitions) { + this.batchErrorHandler.clearThreadStateFor(partitions); + this.recordErrorHandler.clearThreadStateFor(partitions); + } + @Override public boolean isAckAfterHandle() { return this.recordErrorHandler.isAckAfterHandle(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java index 95c59de154..a7694e77ad 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java @@ -16,12 +16,14 @@ package org.springframework.kafka.listener; +import java.util.Collection; import java.util.List; import java.util.function.BiConsumer; import java.util.function.BiFunction; import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; import org.jspecify.annotations.Nullable; import org.springframework.core.log.LogAccessor; @@ -179,4 +181,8 @@ public void clearThreadState() { this.failureTracker.clearThreadState(); } + public void clearThreadStateFor(Collection partitions) { + this.failureTracker.clearThreadStateFor(partitions); + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java index af41fe4931..e8a69e73c7 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordTracker.java @@ -246,6 +246,13 @@ void clearThreadState() { this.failures.remove(Thread.currentThread()); } + void clearThreadStateFor(java.util.Collection partitions) { + Map map = this.failures.get(Thread.currentThread()); + if (map != null) { + partitions.forEach(map::remove); + } + } + ConsumerAwareRecordRecoverer getRecoverer() { return this.recoverer; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index f773d0a663..c17e8620da 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -2397,7 +2397,7 @@ private List> createRecordList(final ConsumerRecords if (this.batchFailed) { this.batchFailed = false; if (this.commonErrorHandler != null) { - this.commonErrorHandler.clearThreadState(); + this.commonErrorHandler.clearThreadStateFor(records.partitions()); } getAfterRollbackProcessor().clearThreadState(); }