GH-1331: KMLC - treat auth. exceptions as fatal

Resolves #1331

Authorization errors create a log storm; stop the container for
such exceptions in the same way we do for `NoOffsetForPartitionException`.

**cherry-pick to 2.2.x**

Author: garyrussell

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

@@ -53,6 +53,7 @@
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.AuthorizationException;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
@@ -711,6 +712,11 @@ public void run() {
 					ListenerConsumer.this.logger.error("No offset and no reset policy", nofpe);
 					break;
 				}
+				catch (AuthorizationException ae) {
+					this.fatalError = true;
+					ListenerConsumer.this.logger.error("Authorization Exception", ae);
+					break;
+				}
 				catch (Exception e) {
 					handleConsumerException(e);
 				}
@@ -848,7 +854,7 @@ public void wrapUp() {
 				}
 			}
 			else {
-				this.logger.error("No offset and no reset policy; stopping container");
+				this.logger.error("Fatal consumer exception; stopping container");
 				KafkaMessageListenerContainer.this.stop();
 			}
 			this.monitorTask.cancel(true);

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

@@ -18,9 +18,11 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
@@ -30,6 +32,7 @@
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.errors.GroupAuthorizationException;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.kafka.core.ConsumerFactory;
@@ -72,4 +75,33 @@ public void testCorrectContainerForConsumerError() throws InterruptedException {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void testConsumerExitWhenNotAuthorized() throws InterruptedException {
+		ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
+		final Consumer consumer = mock(Consumer.class);
+		willAnswer(invocation -> {
+			Thread.sleep(100);
+			throw new GroupAuthorizationException("grp");
+		}).given(consumer).poll(any());
+		CountDownLatch latch = new CountDownLatch(1);
+		willAnswer(invocation -> {
+			latch.countDown();
+			return null;
+		}).given(consumer).close();
+		given(consumerFactory.createConsumer(eq("grp"), eq(""), eq("-0"), any()))
+			.willReturn(consumer);
+		ContainerProperties containerProperties = new ContainerProperties("foo");
+		containerProperties.setGroupId("grp");
+		containerProperties.setMessageListener((MessageListener) record -> { });
+		containerProperties.setMissingTopicsFatal(false);
+		containerProperties.setShutdownTimeout(10);
+		ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(consumerFactory,
+				containerProperties);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		verify(consumer).close();
+		container.stop();
+	}
+
 }