GH-1902: AuthenticationException Fatal By Default

Resolves #1902

Rename property and use for both exception types.

* Fix checkstyle.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -1128,12 +1128,12 @@ It does not apply if the container is configured to listen to a topic pattern (r
 Previously, the container threads looped within the `consumer.poll()` method waiting for the topic to appear while logging many messages.
 Aside from the logs, there was no indication that there was a problem.
 
-As of version 2.3.5, a new container property called `authorizationExceptionRetryInterval` has been introduced.
-This causes the container to retry fetching messages after getting any `AuthorizationException` from `KafkaConsumer`.
-This can happen when, for example, the configured user is denied access to read certain topic.
-Defining `authorizationExceptionRetryInterval` should help the application to recover as soon as proper permissions are granted.
+As of version 2.8, a new container property `authExceptionRetryInterval` has been introduced.
+This causes the container to retry fetching messages after getting any `AuthenticationException` or `AuthorizationException` from the `KafkaConsumer`.
+This can happen when, for example, the configured user is denied access to read a certain topic or credentials are incorrect.
+Defining `authExceptionRetryInterval` allows the container to recover when proper permissions are granted.
 
-NOTE: By default, no interval is configured - authorization errors are considered fatal, which causes the container to stop.
+NOTE: By default, no interval is configured - authentication and authorization errors are considered fatal, which causes the container to stop.
 
 Starting with version 2.8, when creating the consumer factory, if you provide deserializers as objects (in the constructor or via the setters), the factory will invoke the `configure()` method to configure them with the configuration properties.
 
@@ -2433,9 +2433,9 @@ When creating the `TopicPartitionOffset` s for the request, only positive, absol
 |Whether or not to commit the initial position on assignment; by default, the initial offset will only be committed if the `ConsumerConfig.AUTO_OFFSET_RESET_CONFIG` is `latest` and it won't run in a transaction even if there is a transaction manager present.
 See the javadocs for `ContainerProperties.AssignmentCommitOption` for more information about the available options.
 
-|[[authorizationExceptionRetryInterval]]<<authorizationExceptionRetryInterval,`authorizationExceptionRetryInterval`>>
+|[[authExceptionRetryInterval]]<<authExceptionRetryInterval,`authExceptionRetryInterval`>>
 |`null`
-|When not null, a `Duration` to sleep between polls when an `AuthorizationException` is thrown by the Kafka client.
+|When not null, a `Duration` to sleep between polls when an `AuthenticationException` or `AuthorizationException` is thrown by the Kafka client.
 When null, such exceptions are considered fatal and the container will stop.
 
 |[[clientId]]<<clientId,`clientId`>>
@@ -2805,7 +2805,7 @@ 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.
+** `AUTH` - an `AuthenticationException` or `AuthorizationException` was thrown and the `authExceptionRetryInterval` 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:

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -61,6 +61,11 @@ See <<error-handlers>> for more information.
 
 The `interceptBeforeTx` container property is now `true` by default.
 
+The `authorizationExceptionRetryInterval` property has been renamed to `authExceptionRetryInterval` and now applies to `AuthenticationException` s in addition to `AuthorizationException` s previously.
+Both exceptions are considered fatal and the container will stop by default, unless this property is set.
+
+See <<kafka-container>> and <<container-props>> for more information.
+
 [[x28-serializers]]
 ==== Serializer/Deserializer Changes
 

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

@@ -23,6 +23,7 @@
 
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.common.errors.AuthenticationException;
 
 import org.springframework.kafka.support.LogIfLevelEnabled;
 import org.springframework.kafka.support.TopicPartitionOffset;
@@ -101,7 +102,7 @@ public class ConsumerProperties {
 
 	private Properties kafkaConsumerProperties = new Properties();
 
-	private Duration authorizationExceptionRetryInterval;
+	private Duration authExceptionRetryInterval;
 
 	private int commitRetries = DEFAULT_COMMIT_RETRIES;
 
@@ -347,22 +348,54 @@ public void setKafkaConsumerProperties(Properties kafkaConsumerProperties) {
 		this.kafkaConsumerProperties = kafkaConsumerProperties;
 	}
 
+	/**
+	 * Get the authentication/authorization retry interval.
+	 * @return the interval.
+	 * @deprecated in favor of {@link #getAuthExceptionRetryInterval()}.
+	 */
+	@Deprecated
+	@Nullable
 	public Duration getAuthorizationExceptionRetryInterval() {
-		return this.authorizationExceptionRetryInterval;
+		return this.authExceptionRetryInterval;
 	}
 
 	/**
-	 * Set the interval between retries after {@code AuthorizationException} is thrown
-	 * by {@code KafkaConsumer}. By default the field is null and retries are disabled.
-	 * In such case the container will be stopped.
+	 * Set the interval between retries after and {@link AuthenticationException} or
+	 * {@code AuthorizationException} is thrown by {@code KafkaConsumer}. By default the
+	 * field is null and retries are disabled. In such case the container will be stopped.
 	 *
 	 * The interval must be less than {@code max.poll.interval.ms} consumer property.
 	 *
 	 * @param authorizationExceptionRetryInterval the duration between retries
 	 * @since 2.3.5
+	 * @deprecated in favor of {@link #setAuthExceptionRetryInterval(Duration)}.
 	 */
+	@Deprecated
 	public void setAuthorizationExceptionRetryInterval(Duration authorizationExceptionRetryInterval) {
-		this.authorizationExceptionRetryInterval = authorizationExceptionRetryInterval;
+		this.authExceptionRetryInterval = authorizationExceptionRetryInterval;
+	}
+
+	/**
+	 * Get the authentication/authorization retry interval.
+	 * @return the interval.
+	 */
+	@Nullable
+	public Duration getAuthExceptionRetryInterval() {
+		return this.authExceptionRetryInterval;
+	}
+
+	/**
+	 * Set the interval between retries after and {@link AuthenticationException} or
+	 * {@code AuthorizationException} is thrown by {@code KafkaConsumer}. By default the
+	 * field is null and retries are disabled. In such case the container will be stopped.
+	 *
+	 * The interval must be less than {@code max.poll.interval.ms} consumer property.
+	 *
+	 * @param authExceptionRetryInterval the duration between retries
+	 * @since 2.8
+	 */
+	public void setAuthExceptionRetryInterval(Duration authExceptionRetryInterval) {
+		this.authExceptionRetryInterval = authExceptionRetryInterval;
 	}
 
 	/**
@@ -449,7 +482,7 @@ protected final String renderProperties() {
 				+ "\n syncCommits=" + this.syncCommits
 				+ (this.syncCommitTimeout != null ? "\n syncCommitTimeout=" + this.syncCommitTimeout : "")
 				+ (this.kafkaConsumerProperties.size() > 0 ? "\n properties=" + this.kafkaConsumerProperties : "")
-				+ "\n authorizationExceptionRetryInterval=" + this.authorizationExceptionRetryInterval
+				+ "\n authExceptionRetryInterval=" + this.authExceptionRetryInterval
 				+ "\n commitRetries=" + this.commitRetries
 				+ "\n fixTxOffsets" + this.fixTxOffsets;
 	}

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

@@ -58,6 +58,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.AuthenticationException;
 import org.apache.kafka.common.errors.AuthorizationException;
 import org.apache.kafka.common.errors.FencedInstanceIdException;
 import org.apache.kafka.common.errors.ProducerFencedException;
@@ -488,7 +489,7 @@ private void publishConsumerStoppedEvent(@Nullable Throwable throwable) {
 			else if (throwable instanceof StopAfterFenceException || throwable instanceof FencedInstanceIdException) {
 				reason = Reason.FENCED;
 			}
-			else if (throwable instanceof AuthorizationException) {
+			else if (throwable instanceof AuthenticationException || throwable instanceof AuthorizationException) {
 				reason = Reason.AUTH;
 			}
 			else if (throwable instanceof NoOffsetForPartitionException) {
@@ -669,8 +670,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final boolean subBatchPerPartition;
 
-		private final Duration authorizationExceptionRetryInterval =
-				this.containerProperties.getAuthorizationExceptionRetryInterval();
+		private final Duration authExceptionRetryInterval =
+				this.containerProperties.getAuthExceptionRetryInterval();
 
 		private final AssignmentCommitOption autoCommitOption = this.containerProperties.getAssignmentCommitOption();
 
@@ -1238,20 +1239,23 @@ public void run() {
 					exitThrowable = nofpe;
 					break;
 				}
-				catch (AuthorizationException ae) {
-					if (this.authorizationExceptionRetryInterval == null) {
-						ListenerConsumer.this.logger.error(ae, "Authorization Exception and no authorizationExceptionRetryInterval set");
+				catch (AuthenticationException | AuthorizationException ae) {
+					if (this.authExceptionRetryInterval == null) {
+						ListenerConsumer.this.logger.error(ae,
+								"Authentcation/Authorization Exception and no authExceptionRetryInterval set");
 						this.fatalError = true;
 						exitThrowable = ae;
 						break;
 					}
 					else {
-						ListenerConsumer.this.logger.error(ae, "Authorization Exception, retrying in " + this.authorizationExceptionRetryInterval.toMillis() + " ms");
+						ListenerConsumer.this.logger.error(ae,
+								"Authentcation/Authorization Exception, retrying in "
+										+ this.authExceptionRetryInterval.toMillis() + " ms");
 						// We can't pause/resume here, as KafkaConsumer doesn't take pausing
 						// into account when committing, hence risk of being flooded with
 						// GroupAuthorizationExceptions.
 						// see: https://github.com/spring-projects/spring-kafka/pull/1337
-						sleepFor(this.authorizationExceptionRetryInterval);
+						sleepFor(this.authExceptionRetryInterval);
 					}
 				}
 				catch (FencedInstanceIdException fie) {

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

@@ -74,6 +74,7 @@
 import org.apache.kafka.clients.consumer.RetriableCommitFailedException;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.AuthenticationException;
 import org.apache.kafka.common.errors.AuthorizationException;
 import org.apache.kafka.common.errors.FencedInstanceIdException;
 import org.apache.kafka.common.errors.RebalanceInProgressException;
@@ -3029,6 +3030,44 @@ public void testCommitErrorHandlerCalled() throws Exception {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void testFatalErrorOnAuthenticationException() throws Exception {
+		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);
+
+		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();
+		try {
+			assertThat(stopped.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(reason.get()).isEqualTo(Reason.AUTH);
+		}
+		finally {
+			container.stop();
+		}
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
 	void testFatalErrorOnAuthorizationException() throws Exception {
@@ -3080,7 +3119,7 @@ void testNotFatalErrorOnAuthorizationException() throws Exception {
 		containerProps.setGroupId("grp");
 		containerProps.setClientId("clientId");
 		containerProps.setMessageListener((MessageListener) r -> { });
-		containerProps.setAuthorizationExceptionRetryInterval(Duration.ofMillis(100));
+		containerProps.setAuthExceptionRetryInterval(Duration.ofMillis(100));
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.start();