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

@@ -29,6 +29,7 @@
 import org.springframework.kafka.support.SeekUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.backoff.BackOff;
+import org.springframework.util.backoff.BackOffExecution;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
@@ -49,6 +50,12 @@
  */
 public class DefaultAfterRollbackProcessor<K, V> extends FailedRecordProcessor implements AfterRollbackProcessor<K, V> {
 
+	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 KafkaTemplate<K, V> kafkaTemplate;
 
 	/**
@@ -119,6 +126,7 @@ public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>,
 
 		// Remove super CTOR when this is removed.
 		super(recoverer, maxFailures);
+		this.backOff = maxFailuresToBackOff(maxFailures);
 	}
 
 	/**
@@ -132,6 +140,7 @@ public DefaultAfterRollbackProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>,
 			BackOff backOff) {
 
 		super(recoverer, backOff);
+		this.backOff = backOff;
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -147,6 +156,10 @@ && isCommitRecovered() && this.kafkaTemplate != null && this.kafkaTemplate.isTra
 					Collections.singletonMap(new TopicPartition(skipped.topic(), skipped.partition()),
 							new OffsetAndMetadata(skipped.offset() + 1)));
 		}
+
+		if (!recoverable && this.backOff != null) {
+			ListenerUtils.unrecoverableBackOff(this.backOff, this.backOffs, this.lastIntervals);
+		}
 	}
 
 	@Override
@@ -196,4 +209,11 @@ public void setKafkaTemplate(KafkaTemplate<K, V> kafkaTemplate) {
 		this.kafkaTemplate = kafkaTemplate;
 	}
 
+	@Override
+	public void clearThreadState() {
+		super.clearThreadState();
+		this.backOffs.remove();
+		this.lastIntervals.remove();
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -71,7 +71,7 @@ protected FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Excep
 		this.classifier = configureDefaultClassifier();
 	}
 
-	private static FixedBackOff maxFailuresToBackOff(int maxFailures) {
+	protected static FixedBackOff maxFailuresToBackOff(int maxFailures) {
 		if (maxFailures < 0) {
 			return new FixedBackOff(0L, FixedBackOff.UNLIMITED_ATTEMPTS);
 		}

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

@@ -603,6 +603,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private Producer<?, ?> producer;
 
+		private boolean batchFailed;
+
 		private volatile boolean consumerPaused;
 
 		private volatile Collection<TopicPartition> assignedPartitions;
@@ -1465,6 +1467,11 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 			try {
 				invokeBatchOnMessage(records, recordList, producer);
 				successTimer(sample);
+				if (this.batchFailed) {
+					this.batchFailed = false;
+					this.batchErrorHandler.clearThreadState();
+					getAfterRollbackProcessor().clearThreadState();
+				}
 			}
 			catch (RuntimeException e) {
 				failureTimer(sample);
@@ -1476,6 +1483,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()) || producer != null) {

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

@@ -32,6 +32,8 @@
 import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.backoff.BackOff;
+import org.springframework.util.backoff.BackOffExecution;
 
 /**
  * Listener utilities.
@@ -131,4 +133,40 @@ public static String recordToString(ConsumerRecord<?, ?> record) {
 		}
 	}
 
+	/**
+	 * 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

@@ -38,7 +38,7 @@ public class SeekToCurrentBatchErrorHandler implements ContainerAwareBatchErrorH
 
 	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;
 
@@ -65,27 +65,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", thrownException);
@@ -94,7 +74,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

@@ -17,10 +17,14 @@
 package org.springframework.kafka.listener;
 
 import static org.assertj.core.api.Assertions.assertThat;
+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;
 
@@ -37,6 +41,9 @@
 
 import org.springframework.kafka.core.KafkaTemplate;
 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
@@ -46,7 +53,7 @@
 public class DefaultAfterRollbackProcessorTests {
 
 	@Test
-	public void testClassifier() {
+	void testClassifier() {
 		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
 		AtomicBoolean recovererShouldFail = new AtomicBoolean(false);
 		DefaultAfterRollbackProcessor<String, String> processor = new DefaultAfterRollbackProcessor<>((r, t) -> {
@@ -86,4 +93,33 @@ public void testClassifier() {
 		inOrder.verifyNoMoreInteractions();
 	}
 
+	@Test
+	void testBatchBackOff() {
+		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
+		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);
+		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);
+		processor.process(records, consumer, illegalState, false);
+		processor.process(records, consumer, illegalState, false);
+		verify(backOff, times(2)).start();
+		verify(execution.get(), times(2)).nextBackOff();
+		processor.clearThreadState();
+		processor.process(records, consumer, illegalState, false);
+		verify(backOff, times(3)).start();
+		verify(recoverer, never()).accept(any(), any());
+	}
+
 }

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" })