GH-2197: Fix Container State After Fatal Stop

Resolves #2197

When a container is stopped due to any fatal error, `isInExpectedState()`
should return false.

**cherry-pick to 2.8.x, 2.7.x**

Author: garyrussell

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

@@ -165,7 +165,6 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 	private String clientIdSuffix;
 
 	private Runnable emergencyStop = () -> stopAbnormally(() -> {
-		// NOSONAR
 	});
 
 	private volatile ListenerConsumer listenerConsumer;
@@ -222,7 +221,7 @@ public KafkaMessageListenerContainer(ConsumerFactory<? super K, ? super V> consu
 	}
 
 	/**
-	 * Set a {@link Runnable} to call whenever an {@link Error} occurs on a listener
+	 * Set a {@link Runnable} to call whenever a fatal error occurs on the listener
 	 * thread.
 	 * @param emergencyStop the Runnable.
 	 * @since 2.2.1
@@ -1275,11 +1274,8 @@ public void run() {
 					exitThrowable = e;
 				}
 				catch (Error e) { // NOSONAR - rethrown
-					Runnable runnable = KafkaMessageListenerContainer.this.emergencyStop;
-					if (runnable != null) {
-						runnable.run();
-					}
 					this.logger.error(e, "Stopping container due to an Error");
+					this.fatalError = true;
 					wrapUp(e);
 					throw e;
 				}
@@ -1740,8 +1736,10 @@ private void wrapUp(@Nullable Throwable throwable) {
 				}
 			}
 			else {
-				this.logger.error("Fatal consumer exception; stopping container");
-				KafkaMessageListenerContainer.this.stop(false);
+				if (!(throwable instanceof Error)) {
+					this.logger.error("Fatal consumer exception; stopping container");
+				}
+				KafkaMessageListenerContainer.this.emergencyStop.run();
 			}
 			this.monitorTask.cancel(true);
 			if (!this.taskSchedulerExplicitlySet) {

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

@@ -99,6 +99,7 @@
 import org.springframework.kafka.event.ConsumerStoppedEvent;
 import org.springframework.kafka.event.ConsumerStoppedEvent.Reason;
 import org.springframework.kafka.event.ConsumerStoppingEvent;
+import org.springframework.kafka.event.ContainerStoppedEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.listener.ContainerProperties.AssignmentCommitOption;
@@ -3068,38 +3069,63 @@ public void handleOtherException(Exception thrownException, Consumer<?, ?> consu
 		container.stop();
 	}
 
-	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
-	void testFatalErrorOnAuthenticationException() throws Exception {
+	void testFatalErrorOnAuthenticationException() throws InterruptedException {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
-		Consumer<Integer, String> consumer = mock(Consumer.class);
-		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
-		given(cf.getConfigurationProperties()).willReturn(new HashMap<>());
-
-		willThrow(AuthenticationException.class)
-				.given(consumer).poll(any());
-
 		ContainerProperties containerProps = new ContainerProperties(topic1);
 		containerProps.setGroupId("grp");
 		containerProps.setClientId("clientId");
 		containerProps.setMessageListener((MessageListener) r -> { });
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
+		testFatalErrorOnAuthenticationException(container, cf);
+	}
+
+	@Test
+	void testFatalErrorOnAuthenticationExceptionConcurrent() throws InterruptedException {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		ContainerProperties containerProps = new ContainerProperties(topic1);
+		containerProps.setGroupId("grp");
+		containerProps.setClientId("clientId");
+		containerProps.setMessageListener((MessageListener) r -> { });
+		ConcurrentMessageListenerContainer<Integer, String> container =
+				new ConcurrentMessageListenerContainer<>(cf, containerProps);
+		testFatalErrorOnAuthenticationException(container, cf);
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private void testFatalErrorOnAuthenticationException(AbstractMessageListenerContainer container,
+			ConsumerFactory<Integer, String> cf) throws InterruptedException {
+
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"),
+				container instanceof ConcurrentMessageListenerContainer ? eq("-0") : isNull(), any()))
+						.willReturn(consumer);
+		given(cf.getConfigurationProperties()).willReturn(new HashMap<>());
+
+		willThrow(AuthenticationException.class)
+				.given(consumer).poll(any());
 
 		AtomicReference<ConsumerStoppedEvent.Reason> reason = new AtomicReference<>();
-		CountDownLatch stopped = new CountDownLatch(1);
+		CountDownLatch consumerStopped = new CountDownLatch(1);
+		CountDownLatch containerStopped = new CountDownLatch(1);
 
 		container.setApplicationEventPublisher(e -> {
 			if (e instanceof ConsumerStoppedEvent) {
 				reason.set(((ConsumerStoppedEvent) e).getReason());
-				stopped.countDown();
+				consumerStopped.countDown();
+			}
+			else if (e instanceof ContainerStoppedEvent) {
+				containerStopped.countDown();
 			}
 		});
 
 		container.start();
 		try {
-			assertThat(stopped.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(consumerStopped.await(10, TimeUnit.SECONDS)).isTrue();
 			assertThat(reason.get()).isEqualTo(Reason.AUTH);
+			assertThat(containerStopped.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(container.isInExpectedState()).isFalse();
 		}
 		finally {
 			container.stop();
@@ -3125,18 +3151,23 @@ void testFatalErrorOnAuthorizationException() throws Exception {
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 
 		AtomicReference<ConsumerStoppedEvent.Reason> reason = new AtomicReference<>();
-		CountDownLatch stopped = new CountDownLatch(1);
+		CountDownLatch consumerStopped = new CountDownLatch(1);
+		CountDownLatch containerStopped = new CountDownLatch(1);
 
 		container.setApplicationEventPublisher(e -> {
 			if (e instanceof ConsumerStoppedEvent) {
 				reason.set(((ConsumerStoppedEvent) e).getReason());
-				stopped.countDown();
+				consumerStopped.countDown();
+			}
+			else if (e instanceof ContainerStoppedEvent) {
+				containerStopped.countDown();
 			}
 		});
 
 		container.start();
-		assertThat(stopped.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(consumerStopped.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(reason.get()).isEqualTo(Reason.AUTH);
+		assertThat(container.isInExpectedState()).isFalse();
 		container.stop();
 	}
 
@@ -3163,6 +3194,7 @@ void testNotFatalErrorOnAuthorizationException() throws Exception {
 		container.start();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		container.stop();
+		assertThat(container.isInExpectedState()).isTrue();
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -3186,13 +3218,14 @@ void testFatalErrorOnFencedInstanceException() throws Exception {
 		CountDownLatch stopped = new CountDownLatch(1);
 
 		container.setApplicationEventPublisher(e -> {
-			if (e instanceof ConsumerStoppedEvent) {
+			if (e instanceof ContainerStoppedEvent) {
 				stopped.countDown();
 			}
 		});
 
 		container.start();
 		assertThat(stopped.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(container.isInExpectedState()).isFalse();
 		container.stop();
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-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.
@@ -127,6 +127,7 @@ public void testOOMKMLC() throws Exception {
 		container.start();
 		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(container.isRunning()).isFalse();
+		assertThat(container.isInExpectedState()).isFalse();
 	}
 
 }