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(); }