GH-1591: Apply BackOff in DARP with Batch Listener

Resolves #1591

`BackOff` was ignored for batch listeners (not recoverable).

- move code from `SeekToCurrentBatchErrorHandler` to `ListenerUtils`
- call from both STCEH and DARP
- also clear the thread state (in both) if a batch fails and subsequently succeeds

**I will do the backports - conflicts are expected**

Author: garyrussell

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

@@ -32,6 +32,7 @@
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.backoff.BackOff;
+import org.springframework.util.backoff.BackOffExecution;
 
 /**
  * Default implementation of {@link AfterRollbackProcessor}. Seeks all
@@ -52,6 +53,12 @@
 public class DefaultAfterRollbackProcessor<K, V> extends FailedRecordProcessor
 		implements AfterRollbackProcessor<K, V>, InitializingBean {
 
+	private final ThreadLocal<BackOffExecution> backOffs = new ThreadLocal<>(); // Intentionally not static
+
+	private final ThreadLocal<Long> lastIntervals = new ThreadLocal<>(); // Intentionally not static
+
+	private final BackOff backOff;
+
 	private KafkaOperations<?, ?> kafkaTemplate;
 
 	/**
@@ -115,6 +122,7 @@ public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>,
 		this.kafkaTemplate = kafkaOperations;
 		super.setCommitRecovered(commitRecovered);
 		checkConfig();
+		this.backOff = backOff;
 	}
 
 	@Override
@@ -156,6 +164,10 @@ && isCommitRecovered() && this.kafkaTemplate.isTransactional()) {
 								new OffsetAndMetadata(skipped.offset() + 1)), consumer.groupMetadata());
 			}
 		}
+
+		if (!recoverable && this.backOff != null) {
+			ListenerUtils.unrecoverableBackOff(this.backOff, this.backOffs, this.lastIntervals);
+		}
 	}
 
 	@Override
@@ -210,4 +222,11 @@ public void setKafkaOperations(KafkaOperations<K, V> kafkaOperations) {
 		this.kafkaTemplate = kafkaOperations;
 	}
 
+	@Override
+	public void clearThreadState() {
+		super.clearThreadState();
+		this.backOffs.remove();
+		this.lastIntervals.remove();
+	}
+
 }

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

@@ -53,7 +53,6 @@ default void handle(Exception thrownException, T data, Consumer<?, ?> consumer)
 	 * @since 2.3
 	 */
 	default void clearThreadState() {
-		// NOSONAR
 	}
 
 	/**

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

@@ -591,6 +591,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private boolean commitRecovered;
 
+		private boolean batchFailed;
+
 		private volatile boolean consumerPaused;
 
 		private volatile Thread consumerThread;
@@ -1585,6 +1587,11 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 			try {
 				invokeBatchOnMessage(records, recordList);
 				successTimer(sample);
+				if (this.batchFailed) {
+					this.batchFailed = false;
+					this.batchErrorHandler.clearThreadState();
+					getAfterRollbackProcessor().clearThreadState();
+				}
 			}
 			catch (RuntimeException e) {
 				failureTimer(sample);
@@ -1596,6 +1603,7 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 					throw e;
 				}
 				try {
+					this.batchFailed = true;
 					invokeBatchErrorHandler(records, recordList, e);
 					// unlikely, but possible, that a batch error handler "handles" the error
 					if ((!acked && !this.autoCommit && this.batchErrorHandler.isAckAfterHandle())

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

@@ -29,6 +29,8 @@
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.backoff.BackOff;
+import org.springframework.util.backoff.BackOffExecution;
 
 /**
  * Listener utilities.
@@ -142,5 +144,41 @@ public static String recordToString(ConsumerRecord<?, ?> record, boolean meta) {
 		}
 	}
 
+	/**
+	 * Sleep according to the {@link BackOff}; when the {@link BackOffExecution} returns
+	 * {@link BackOffExecution#STOP} sleep for the previous backOff.
+	 * @param backOff the {@link BackOff} to create a new {@link BackOffExecution}.
+	 * @param executions a thread local containing the {@link BackOffExecution} for this
+	 * thread.
+	 * @param lastIntervals a thread local containing the previous {@link BackOff}
+	 * interval for this thread.
+	 * @since 2.3.12
+	 */
+	public static void unrecoverableBackOff(BackOff backOff, ThreadLocal<BackOffExecution> executions,
+			ThreadLocal<Long> lastIntervals) {
+
+		BackOffExecution backOffExecution = executions.get();
+		if (backOffExecution == null) {
+			backOffExecution = backOff.start();
+			executions.set(backOffExecution);
+		}
+		Long interval = backOffExecution.nextBackOff();
+		if (interval == BackOffExecution.STOP) {
+			interval = lastIntervals.get();
+			if (interval == null) {
+				interval = Long.valueOf(0);
+			}
+		}
+		lastIntervals.set(interval);
+		if (interval > 0) {
+			try {
+				Thread.sleep(interval);
+			}
+			catch (@SuppressWarnings("unused") InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+		}
+	}
+
 }
 

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

@@ -39,7 +39,7 @@ public class SeekToCurrentBatchErrorHandler extends KafkaExceptionLogLevelAware
 
 	private final ThreadLocal<BackOffExecution> backOffs = new ThreadLocal<>(); // Intentionally not static
 
-	private final ThreadLocal<Long> lastInterval = new ThreadLocal<>(); // Intentionally not static
+	private final ThreadLocal<Long> lastIntervals = new ThreadLocal<>(); // Intentionally not static
 
 	private BackOff backOff;
 
@@ -66,27 +66,7 @@ public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consum
 				.forEach(consumer::seek);
 
 		if (this.backOff != null) {
-			BackOffExecution backOffExecution = this.backOffs.get();
-			if (backOffExecution == null) {
-				backOffExecution = this.backOff.start();
-				this.backOffs.set(backOffExecution);
-			}
-			Long interval = backOffExecution.nextBackOff();
-			if (interval == BackOffExecution.STOP) {
-				interval = this.lastInterval.get();
-				if (interval == null) {
-					interval = Long.valueOf(0);
-				}
-			}
-			this.lastInterval.set(interval);
-			if (interval > 0) {
-				try {
-					Thread.sleep(interval);
-				}
-				catch (@SuppressWarnings("unused") InterruptedException e) {
-					Thread.currentThread().interrupt();
-				}
-			}
+			ListenerUtils.unrecoverableBackOff(this.backOff, this.backOffs, this.lastIntervals);
 		}
 
 		throw new KafkaException("Seek to current after exception", getLogLevel(), thrownException);
@@ -95,7 +75,7 @@ public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consum
 	@Override
 	public void clearThreadState() {
 		this.backOffs.remove();
-		this.lastInterval.remove();
+		this.lastIntervals.remove();
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessorTests.java

@@ -20,9 +20,11 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -41,6 +43,9 @@
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.listener.ContainerProperties.EOSMode;
 import org.springframework.kafka.support.serializer.DeserializationException;
+import org.springframework.util.backoff.BackOff;
+import org.springframework.util.backoff.BackOffExecution;
+import org.springframework.util.backoff.FixedBackOff;
 
 /**
  * @author Gary Russell
@@ -50,7 +55,7 @@
 public class DefaultAfterRollbackProcessorTests {
 
 	@Test
-	public void testClassifier() {
+	void testClassifier() {
 		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
 		AtomicBoolean recovererShouldFail = new AtomicBoolean(false);
 		@SuppressWarnings("unchecked")
@@ -92,4 +97,36 @@ public void testClassifier() {
 		inOrder.verifyNoMoreInteractions();
 	}
 
+	@Test
+	void testBatchBackOff() {
+		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
+		@SuppressWarnings("unchecked")
+		KafkaOperations<String, String> template = mock(KafkaOperations.class);
+		given(template.isTransactional()).willReturn(true);
+		BackOff backOff = spy(new FixedBackOff(0, 1));
+		AtomicReference<BackOffExecution> execution = new AtomicReference<>();
+		willAnswer(inv -> {
+			BackOffExecution exec = spy((BackOffExecution) inv.callRealMethod());
+			execution.set(exec);
+			return exec;
+		}).given(backOff).start();
+		ConsumerRecordRecoverer recoverer = mock(ConsumerRecordRecoverer.class);
+		DefaultAfterRollbackProcessor<String, String> processor = new DefaultAfterRollbackProcessor<>(recoverer,
+				backOff, template, false);
+		ConsumerRecord<String, String> record1 = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
+		ConsumerRecord<String, String> record2 = new ConsumerRecord<>("foo", 1, 1L, "foo", "bar");
+		List<ConsumerRecord<String, String>> records = Arrays.asList(record1, record2);
+		IllegalStateException illegalState = new IllegalStateException();
+		@SuppressWarnings("unchecked")
+		Consumer<String, String> consumer = mock(Consumer.class);
+		given(consumer.groupMetadata()).willReturn(new ConsumerGroupMetadata("foo"));
+		processor.process(records, consumer, illegalState, false, EOSMode.BETA);
+		processor.process(records, consumer, illegalState, false, EOSMode.BETA);
+		verify(backOff, times(2)).start();
+		verify(execution.get(), times(2)).nextBackOff();
+		processor.clearThreadState();
+		processor.process(records, consumer, illegalState, false, EOSMode.BETA);
+		verify(backOff, times(3)).start();
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentBatchErrorHandlerTests.java

@@ -26,6 +26,8 @@
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
 
 import java.time.Duration;
 import java.util.Arrays;
@@ -65,6 +67,7 @@
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
@@ -133,10 +136,17 @@ void testBackOff() {
 		long t1 = System.currentTimeMillis();
 		for (int i = 0; i < 10; i++) {
 			assertThatThrownBy(() -> eh.handle(ex, crs, mock(Consumer.class), mock(MessageListenerContainer.class)))
-				.isInstanceOf(KafkaException.class)
-				.hasCause(ex);
+					.isInstanceOf(KafkaException.class)
+					.hasCause(ex);
 		}
 		assertThat(System.currentTimeMillis() - t1).isGreaterThanOrEqualTo(100L);
+		eh.clearThreadState();
+		BackOff backOff = spy(new FixedBackOff(0L, 0L));
+		eh.setBackOff(backOff);
+		assertThatThrownBy(() -> eh.handle(ex, crs, mock(Consumer.class), mock(MessageListenerContainer.class)))
+				.isInstanceOf(KafkaException.class)
+				.hasCause(ex);
+		verify(backOff).start();
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })