Fix batch listener backoff state reset in multi-partition containers

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<TopicPartition> partitions) {
+		this.defaultErrorHandler.clearThreadStateFor(partitions);
+		this.delegates.values().forEach(h -> h.clearThreadStateFor(partitions));
+	}
+
 	@Override
 	public boolean isAckAfterHandle() {
 		return this.defaultErrorHandler.isAckAfterHandle();

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<TopicPartition> partitions) {
+	}
+
 	/**
 	 * Return true if the offset should be committed for a handled error (no exception
 	 * thrown).

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<TopicPartition> partitions) {
+		this.batchErrorHandler.clearThreadStateFor(partitions);
+		this.recordErrorHandler.clearThreadStateFor(partitions);
+	}
+
 	@Override
 	public boolean isAckAfterHandle() {
 		return this.recordErrorHandler.isAckAfterHandle();

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<TopicPartition> partitions) {
+		this.failureTracker.clearThreadStateFor(partitions);
+	}
+
 }

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<TopicPartition> partitions) {
+		Map<TopicPartition, FailedRecord> map = this.failures.get(Thread.currentThread());
+		if (map != null) {
+			partitions.forEach(map::remove);
+		}
+	}
+
 	ConsumerAwareRecordRecoverer getRecoverer() {
 		return this.recoverer;
 	}

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -2397,7 +2397,7 @@ private List<ConsumerRecord<K, V>> createRecordList(final ConsumerRecords<K, V>
 				if (this.batchFailed) {
 					this.batchFailed = false;
 					if (this.commonErrorHandler != null) {
-						this.commonErrorHandler.clearThreadState();
+						this.commonErrorHandler.clearThreadStateFor(records.partitions());
 					}
 					getAfterRollbackProcessor().clearThreadState();
 				}