GH-2382: Fix RetryingBatchErrorHandler Cross Talk

Resolves #2382

Change `retrying` boolean to `ThreadLocal<Boolean>`.

Author: garyrussell

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

@@ -55,9 +55,9 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware
 	@SuppressWarnings("deprecation")
 	private final CommonErrorHandler seeker = new ErrorHandlerAdapter(new SeekToCurrentBatchErrorHandler());
 
-	private boolean ackAfterHandle = true;
+	private final ThreadLocal<Boolean> retrying = ThreadLocal.withInitial(() -> false);
 
-	private boolean retrying;
+	private boolean ackAfterHandle = true;
 
 	/**
 	 * Construct an instance with a default {@link FixedBackOff} (unlimited attempts with
@@ -104,14 +104,18 @@ public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> re
 			this.logger.error(thrownException, "Called with no records; consumer exception");
 			return;
 		}
-		this.retrying = true;
-		ErrorHandlingUtils.retryBatch(thrownException, records, consumer, container, invokeListener, this.backOff,
-				this.seeker, this.recoverer, this.logger, getLogLevel());
-		this.retrying = false;
+		this.retrying.set(true);
+		try {
+			ErrorHandlingUtils.retryBatch(thrownException, records, consumer, container, invokeListener, this.backOff,
+					this.seeker, this.recoverer, this.logger, getLogLevel());
+		}
+		finally {
+			this.retrying.set(false);
+		}
 	}
 
 	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
-		if (this.retrying) {
+		if (this.retrying.get()) {
 			consumer.pause(consumer.assignment());
 		}
 	}

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

@@ -16,17 +16,21 @@
 
 package org.springframework.kafka.listener;
 
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 
+import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -42,6 +46,7 @@
 import org.mockito.InOrder;
 
 import org.springframework.kafka.KafkaException;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
@@ -205,4 +210,35 @@ void rePauseOnRebalance() {
 		verifyNoMoreInteractions(consumer);
 	}
 
+	@Test
+	void resetRetryingFlagOnExceptionFromRetryBatch() {
+		FallbackBatchErrorHandler eh = new FallbackBatchErrorHandler(new FixedBackOff(0L, 1L), (consumerRecord, e) -> { });
+
+		Consumer<?, ?> consumer = mock(Consumer.class);
+		// KafkaException could be thrown from SeekToCurrentBatchErrorHandler, but it is hard to mock
+		KafkaException exception = new KafkaException("Failed consumer.resume()");
+		willThrow(exception).given(consumer).resume(any());
+
+		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		given(container.isRunning()).willReturn(true);
+
+		Map<TopicPartition, List<ConsumerRecord<Object, Object>>> map = new HashMap<>();
+		map.put(new TopicPartition("foo", 0),
+				Collections.singletonList(new ConsumerRecord<>("foo", 0, 0L, "foo", "bar")));
+		ConsumerRecords<?, ?> records = new ConsumerRecords<>(map);
+
+		assertThatThrownBy(() -> eh.handle(new RuntimeException(), records, consumer, container, () -> { }))
+				.isSameAs(exception);
+
+		assertThat(getRetryingFieldValue(eh))
+				.withFailMessage("retrying field was not reset to false")
+				.isFalse();
+	}
+
+	private boolean getRetryingFieldValue(FallbackBatchErrorHandler errorHandler) {
+		Field field = ReflectionUtils.findField(FallbackBatchErrorHandler.class, "retrying");
+		ReflectionUtils.makeAccessible(field);
+		ThreadLocal<Boolean> value = (ThreadLocal<Boolean>) ReflectionUtils.getField(field, errorHandler);
+		return value.get();
+	}
 }