GH-2179: Exit RetryingBatchErrorHandler on Stop

Resolves #2179

Backported.

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -115,6 +115,9 @@ public void handle(Exception thrownException, ConsumerRecords<?, ?> records,
 					this.seeker.handle(thrownException, records, consumer, container);
 					throw new KafkaException("Interrupted during retry", getLogLevel(), e1);
 				}
+				if (!container.isRunning()) {
+					throw new KafkaException("Container stopped during retries");
+				}
 				try {
 					invokeListener.run();
 					return;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -19,6 +19,8 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -29,13 +31,15 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 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.junit.jupiter.api.Test;
 
+import org.springframework.kafka.KafkaException;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
@@ -62,6 +66,7 @@ void recover() {
 		ConsumerRecords<?, ?> records = new ConsumerRecords<>(map);
 		Consumer<?, ?> consumer = mock(Consumer.class);
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		given(container.isRunning()).willReturn(true);
 		eh.handle(new RuntimeException(), records, consumer, container, () -> {
 			this.invoked++;
 			throw new RuntimeException();
@@ -90,6 +95,7 @@ void successOnRetry() {
 		ConsumerRecords<?, ?> records = new ConsumerRecords<>(map);
 		Consumer<?, ?> consumer = mock(Consumer.class);
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		given(container.isRunning()).willReturn(true);
 		eh.handle(new RuntimeException(), records, consumer, container, () -> this.invoked++);
 		assertThat(this.invoked).isEqualTo(1);
 		assertThat(recovered).hasSize(0);
@@ -116,6 +122,7 @@ void recoveryFails() {
 		ConsumerRecords<?, ?> records = new ConsumerRecords<>(map);
 		Consumer<?, ?> consumer = mock(Consumer.class);
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		given(container.isRunning()).willReturn(true);
 		assertThatExceptionOfType(RuntimeException.class).isThrownBy(() ->
 		eh.handle(new RuntimeException(), records, consumer, container, () -> {
 			this.invoked++;
@@ -131,4 +138,31 @@ void recoveryFails() {
 		verify(consumer).seek(new TopicPartition("foo", 1), 0L);
 	}
 
+	@Test
+	void exitOnContainerStop() {
+		this.invoked = 0;
+		List<ConsumerRecord<?, ?>> recovered = new ArrayList<>();
+		RetryingBatchErrorHandler eh = new RetryingBatchErrorHandler(new FixedBackOff(0, 99999), (cr, ex) ->  {
+			recovered.add(cr);
+		});
+		Map<TopicPartition, List<ConsumerRecord<Object, Object>>> map = new HashMap<>();
+		map.put(new TopicPartition("foo", 0),
+				Collections.singletonList(new ConsumerRecord<>("foo", 0, 0L, "foo", "bar")));
+		map.put(new TopicPartition("foo", 1),
+				Collections.singletonList(new ConsumerRecord<>("foo", 1, 0L, "foo", "bar")));
+		ConsumerRecords<?, ?> records = new ConsumerRecords<>(map);
+		Consumer<?, ?> consumer = mock(Consumer.class);
+		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		AtomicBoolean stopped = new AtomicBoolean(true);
+		willAnswer(inv -> stopped.get()).given(container).isRunning();
+		assertThatExceptionOfType(KafkaException.class).isThrownBy(() ->
+			eh.handle(new RuntimeException(), records, consumer, container, () -> {
+				this.invoked++;
+				stopped.set(false);
+				throw new RuntimeException();
+			})
+		).withMessage("Container stopped during retries");
+		assertThat(this.invoked).isEqualTo(1);
+	}
+
 }