GH-1336: Introduce authExceptionRetryInterval

Resolves #1336

- to retry on auth errors

Author: lukasz-kaminski

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

@@ -97,6 +97,8 @@ public class ConsumerProperties {
 
 	private Properties kafkaConsumerProperties = new Properties();
 
+	private Duration authorizationExceptionRetryInterval;
+
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -304,6 +306,24 @@ public void setKafkaConsumerProperties(Properties kafkaConsumerProperties) {
 		this.kafkaConsumerProperties = kafkaConsumerProperties;
 	}
 
+	public Duration getAuthorizationExceptionRetryInterval() {
+		return this.authorizationExceptionRetryInterval;
+	}
+
+	/**
+	 * 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.
+	 *
+	 * The interval must be less than {@code max.poll.interval.ms} consumer property.
+	 *
+	 * @param authorizationExceptionRetryInterval the duration between retries
+	 * @since 2.3.5
+	 */
+	public void setAuthorizationExceptionRetryInterval(Duration authorizationExceptionRetryInterval) {
+		this.authorizationExceptionRetryInterval = authorizationExceptionRetryInterval;
+	}
+
 	@Override
 	public String toString() {
 		return "ConsumerProperties ["
@@ -322,7 +342,8 @@ protected final String renderProperties() {
 				+ (this.commitCallback != null ? ", commitCallback=" + this.commitCallback : "")
 				+ ", syncCommits=" + this.syncCommits
 				+ (this.syncCommitTimeout != null ? ", syncCommitTimeout=" + this.syncCommitTimeout : "")
-				+ (this.kafkaConsumerProperties.size() > 0 ? ", properties=" + this.kafkaConsumerProperties : "");
+				+ (this.kafkaConsumerProperties.size() > 0 ? ", properties=" + this.kafkaConsumerProperties : ""
+				+ ", authorizationExceptionRetryInterval=" + this.authorizationExceptionRetryInterval);
 	}
 
 	private String renderTopics() {

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

@@ -38,6 +38,7 @@
  * @author Artem Bilan
  * @author Artem Yakshin
  * @author Johnny Lim
+ * @author Lukasz Kaminski
  */
 public class ContainerProperties extends ConsumerProperties {
 
@@ -600,7 +601,7 @@ public boolean isSubBatchPerPartition() {
 	 * When using a batch message listener whether to dispatch records by partition (with
 	 * a transaction for each sub batch if transactions are in use) or the complete batch
 	 * received by the {@code poll()}. Useful when using transactions to enable zombie
-	 * fencing, by using a {code transactional.id} that is unique for each
+	 * fencing, by using a {@code transactional.id} that is unique for each
 	 * group/topic/partition.
 	 * @param subBatchPerPartition true for a separate transaction for each partition.
 	 * @since 2.3.2

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

@@ -122,6 +122,7 @@
  * @author Chen Binbin
  * @author Yang Qiju
  * @author Tom van den Berge
+ * @author Lukasz Kaminski
  */
 public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		extends AbstractMessageListenerContainer<K, V> {
@@ -568,6 +569,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private Map<TopicPartition, OffsetMetadata> definedPartitions;
 
+		private Duration authorizationExceptionRetryInterval = this.containerProperties.getAuthorizationExceptionRetryInterval();
+
 		private int count;
 
 		private long last = System.currentTimeMillis();
@@ -910,9 +913,19 @@ public void run() {
 					break;
 				}
 				catch (AuthorizationException ae) {
-					this.fatalError = true;
-					ListenerConsumer.this.logger.error(ae, "Authorization Exception");
-					break;
+					if (this.authorizationExceptionRetryInterval == null) {
+						ListenerConsumer.this.logger.error(ae, "Authorization Exception and no authorizationExceptionRetryInterval set");
+						this.fatalError = true;
+						break;
+					}
+					else {
+						ListenerConsumer.this.logger.error(ae, "Authorization Exception, retrying in " + this.authorizationExceptionRetryInterval.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);
+					}
 				}
 				catch (Exception e) {
 					handleConsumerException(e);
@@ -949,7 +962,7 @@ protected void pollAndInvoke() {
 			if (this.seeks.size() > 0) {
 				processSeeks();
 			}
-			checkPaused();
+			pauseConsumerIfNecessary();
 			this.lastPoll = System.currentTimeMillis();
 			this.polling.set(true);
 			ConsumerRecords<K, V> records = doPoll();
@@ -963,7 +976,7 @@ protected void pollAndInvoke() {
 				}
 				return;
 			}
-			checkResumed();
+			resumeConsumerIfNeccessary();
 			debugRecords(records);
 			if (records != null && records.count() > 0) {
 				if (this.containerProperties.getIdleEventInterval() != null) {
@@ -1020,7 +1033,16 @@ private void debugRecords(ConsumerRecords<K, V> records) {
 			}
 		}
 
-		private void checkPaused() {
+		private void sleepFor(Duration duration) {
+			try {
+				TimeUnit.MILLISECONDS.sleep(duration.toMillis());
+			}
+			catch (InterruptedException e) {
+				this.logger.error(e, "Interrupted while sleeping");
+			}
+		}
+
+		private void pauseConsumerIfNecessary() {
 			if (!this.consumerPaused && isPaused()) {
 				this.consumer.pause(this.consumer.assignment());
 				this.consumerPaused = true;
@@ -1029,7 +1051,7 @@ private void checkPaused() {
 			}
 		}
 
-		private void checkResumed() {
+		private void resumeConsumerIfNeccessary() {
 			if (this.consumerPaused && !isPaused()) {
 				this.logger.debug(() -> "Resuming consumption from: " + this.consumer.paused());
 				Set<TopicPartition> paused = this.consumer.paused();

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

@@ -25,6 +25,7 @@
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -66,6 +67,7 @@
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.producer.ProducerConfig;
 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.serialization.IntegerDeserializer;
 import org.junit.jupiter.api.BeforeAll;
@@ -110,6 +112,7 @@
  * @author Martin Dam
  * @author Artem Bilan
  * @author Loic Talhouarne
+ * @author Lukasz Kaminski
  */
 @EmbeddedKafka(topics = { KafkaMessageListenerContainerTests.topic1, KafkaMessageListenerContainerTests.topic2,
 		KafkaMessageListenerContainerTests.topic3, KafkaMessageListenerContainerTests.topic4,
@@ -2655,6 +2658,37 @@ public void testCommitErrorHandlerCalled() throws Exception {
 		container.stop();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void testFatalErrorOnAuthorizationException() 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(AuthorizationException.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);
+
+		CountDownLatch stopping = new CountDownLatch(1);
+
+		container.setApplicationEventPublisher(e -> {
+			if (e instanceof ConsumerStoppingEvent) {
+				stopping.countDown();
+			}
+		});
+
+		container.start();
+		assertThat(stopping.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+	}
+
 	private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
 		Consumer<?, ?> consumer = spy(
 				KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class));

src/reference/asciidoc/kafka.adoc

@@ -757,6 +757,13 @@ 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.
+
+NOTE: By default, no interval is configured - authorization errors are considered fatal, which causes the container to stop.
+
 [[using-ConcurrentMessageListenerContainer]]
 ====== Using `ConcurrentMessageListenerContainer`
 

src/reference/asciidoc/whats-new.adoc

@@ -33,6 +33,11 @@ It is now called after a timeout (as well as when records arrive); the second pa
 
 See <<aggregating-request-reply>> for more information.
 
+==== Listener Container
+
+The `ContainerProperties` provides an `authorizationExceptionRetryInterval` option to let the listener container to retry after any `AuthorizationException` is thrown by the `KafkaConsumer`.
+See its JavaDocs and <<kafka-container>> for more information.
+
 === Migration Guide
 
 * This release is essentially the same as the 2.3.x line, except it has been compiled against the 2.4 `kafka-clients` jar, due to a binary incompatibility.