GH-1643: Add More ConsumerStoppedEvent.Reasons

Resolves #1643

**cherry-pick to 2.5.x**

Author: garyrussell

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

@@ -31,6 +31,11 @@ public class ConsumerStoppedEvent extends KafkaEvent {
 
 	private static final long serialVersionUID = 1L;
 
+	/**
+	 * Reasons for stopping a consumer.
+	 * @since 2.5.9
+	 *
+	 */
 	public enum Reason {
 
 		/**
@@ -44,6 +49,18 @@ public enum Reason {
 		 */
 		FENCED,
 
+		/**
+		 * An authorization exception occurred.
+		 * @since 2.5.10
+		 */
+		AUTH,
+
+		/**
+		 * No offset found for a partition and no reset policy.
+		 * @since 2.5.10
+		 */
+		NO_OFFSET,
+
 		/**
 		 * A {@link java.lang.Error} was thrown.
 		 */

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

@@ -396,12 +396,24 @@ private void publishConsumerStoppingEvent(Consumer<?, ?> consumer) {
 
 	private void publishConsumerStoppedEvent(@Nullable Throwable throwable) {
 		if (getApplicationEventPublisher() != null) {
+			Reason reason;
+			if (throwable instanceof Error) {
+				reason = Reason.ERROR;
+			}
+			else if (throwable instanceof StopAfterFenceException || throwable instanceof FencedInstanceIdException) {
+				reason = Reason.FENCED;
+			}
+			else if (throwable instanceof AuthorizationException) {
+				reason = Reason.AUTH;
+			}
+			else if (throwable instanceof NoOffsetForPartitionException) {
+				reason = Reason.NO_OFFSET;
+			}
+			else {
+				reason = Reason.NORMAL;
+			}
 			getApplicationEventPublisher().publishEvent(new ConsumerStoppedEvent(this, this.thisOrParentContainer,
-					throwable instanceof Error
-						? Reason.ERROR
-						: throwable instanceof StopAfterFenceException
-							? Reason.FENCED
-							: Reason.NORMAL));
+					reason));
 		}
 	}
 
@@ -1050,12 +1062,14 @@ public void run() {
 				catch (NoOffsetForPartitionException nofpe) {
 					this.fatalError = true;
 					ListenerConsumer.this.logger.error(nofpe, "No offset and no reset policy");
+					exitThrowable = nofpe;
 					break;
 				}
 				catch (AuthorizationException ae) {
 					if (this.authorizationExceptionRetryInterval == null) {
 						ListenerConsumer.this.logger.error(ae, "Authorization Exception and no authorizationExceptionRetryInterval set");
 						this.fatalError = true;
+						exitThrowable = ae;
 						break;
 					}
 					else {
@@ -1071,6 +1085,7 @@ public void run() {
 					this.fatalError = true;
 					ListenerConsumer.this.logger.error(fie, "'" + ConsumerConfig.GROUP_INSTANCE_ID_CONFIG
 							+ "' has been fenced");
+					exitThrowable = fie;
 					break;
 				}
 				catch (StopAfterFenceException e) {

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

@@ -94,6 +94,7 @@
 import org.springframework.kafka.event.ConsumerPausedEvent;
 import org.springframework.kafka.event.ConsumerResumedEvent;
 import org.springframework.kafka.event.ConsumerStoppedEvent;
+import org.springframework.kafka.event.ConsumerStoppedEvent.Reason;
 import org.springframework.kafka.event.ConsumerStoppingEvent;
 import org.springframework.kafka.event.NonResponsiveConsumerEvent;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
@@ -2726,16 +2727,19 @@ void testFatalErrorOnAuthorizationException() throws Exception {
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 
+		AtomicReference<ConsumerStoppedEvent.Reason> reason = new AtomicReference<>();
 		CountDownLatch stopped = new CountDownLatch(1);
 
 		container.setApplicationEventPublisher(e -> {
 			if (e instanceof ConsumerStoppedEvent) {
+				reason.set(((ConsumerStoppedEvent) e).getReason());
 				stopped.countDown();
 			}
 		});
 
 		container.start();
 		assertThat(stopped.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(reason.get()).isEqualTo(Reason.AUTH);
 		container.stop();
 	}
 

src/reference/asciidoc/kafka.adoc

@@ -2555,6 +2555,8 @@ In addition, the `ConsumerStoppedEvent` has the following additional property:
 ** `NORMAL` - the consumer stopped normally (container was stopped).
 ** `ERROR` - a `java.lang.Error` was thrown.
 ** `FENCED` - the transactional producer was fenced and the `stopContainerWhenFenced` container property is `true`.
+** `AUTH` - an `AuthorizationException` was thrown and the `authorizationExceptionRetryInterval` is not configured.
+** `NO_OFFSET` - there is no offset for a partition and the `auto.offset.reset` policy is `none`.
 
 You can use this event to restart the container after such a condition: