GH-3948: Return ConsumerStoppedEvent with Abnormal reason when MLC stopped abnormally

Fixes #3948

Issue link: #3948

Signed-off-by: Alamuri Lokesh <[email protected]>

Author: AlamuriLokesh

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/events.adoc

@@ -102,6 +102,7 @@ In addition, the `ConsumerStoppedEvent` has the following additional property:
 
 * `reason`:
 ** `NORMAL` - the consumer stopped normally (container was stopped).
+** `ABNORMAL` - the consumer stopped abnormally (container was stopped abnormally).
 ** `ERROR` - a `java.lang.Error` was thrown.
 ** `FENCED` - the transactional producer was fenced and the `stopContainerWhenFenced` container property is `true`.
 ** `AUTH` - an `AuthenticationException` or `AuthorizationException` was thrown and the `authExceptionRetryInterval` is not configured.

spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerStoppedEvent.java

@@ -24,6 +24,7 @@
  * to restart a container that was stopped because a transactional producer was fenced.
  *
  * @author Gary Russell
+ * @author Lokesh Alamuri
  * @since 2.2
  *
  */
@@ -43,6 +44,13 @@ public enum Reason {
 		 */
 		NORMAL,
 
+		/**
+		 * The consumer was stopped because the container was stopped abnormally.
+		 * @since 4.0
+		 *
+		 */
+		ABNORMAL,
+
 		/**
 		 * The transactional producer was fenced and the container
 		 * {@code stopContainerWhenFenced} property is true.

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

@@ -537,6 +537,9 @@ else if (throwable instanceof AuthenticationException || throwable instanceof Au
 			else if (throwable instanceof NoOffsetForPartitionException) {
 				reason = Reason.NO_OFFSET;
 			}
+			else if (!this.isStoppedNormally()) {
+				reason = Reason.ABNORMAL;
+			}
 			else {
 				reason = Reason.NORMAL;
 			}

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

@@ -1088,11 +1088,29 @@ protected Consumer<Integer, String> createKafkaConsumer(String groupId, String c
 		assertThat(container.getContainers()).
 				doesNotContain(childContainer1);
 
-		container.getContainers().forEach(containerForEach -> containerForEach.stop());
+		KafkaMessageListenerContainer<Integer, String> childContainer0SecRun = container.getContainers().get(0);
+		KafkaMessageListenerContainer<Integer, String> childContainer1SecRun = container.getContainers().get(1);
+
+		childContainer0SecRun.stopAbnormally(() -> {
+		});
+
+		childContainer1SecRun.stop();
+
 		assertThat(container.getContainers()).isNotEmpty();
 		container.stop();
 		assertThat(concurrentContainerSecondStopLatch.await(30, TimeUnit.SECONDS)).isTrue();
 
+		events.stream().forEach(event -> {
+			if (event.getContainer(MessageListenerContainer.class).equals(childContainer0SecRun)
+					&& event instanceof ConsumerStoppedEvent) {
+				assertThat(((ConsumerStoppedEvent) event).getReason()).isEqualTo(ConsumerStoppedEvent.Reason.ABNORMAL);
+			}
+			else if (event.getContainer(MessageListenerContainer.class).equals(childContainer1SecRun)
+					&& event instanceof ConsumerStoppedEvent) {
+				assertThat(((ConsumerStoppedEvent) event).getReason()).isEqualTo(ConsumerStoppedEvent.Reason.NORMAL);
+			}
+		});
+
 		this.logger.info("Stop containerStartStop");
 	}