Return ConsumerStoppedEvent with Abnormal reason when MLC stopped Abnormally

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

Author: AlamuriLokesh

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2025 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.
@@ -43,6 +43,11 @@ public enum Reason {
 		 */
 		NORMAL,
 
+		/**
+		 * The consumer was stopped because the container was stopped abnormally.
+		 */
+		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");
 	}