GH-1496: Handle RetriableCommitFailedException

Resolves #1496

Certain commit exceptions are retriable.

**I will back-port as necessary**

* Fix import

Author: garyrussell

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

@@ -44,6 +44,8 @@ public class ConsumerProperties {
 	 */
 	public static final long DEFAULT_POLL_TIMEOUT = 5_000L;
 
+	private static final int DEFAULT_COMMIT_RETRIES = 3;
+
 	/**
 	 * Topic names.
 	 */
@@ -99,6 +101,8 @@ public class ConsumerProperties {
 
 	private Duration authorizationExceptionRetryInterval;
 
+	private int commitRetries = DEFAULT_COMMIT_RETRIES;
+
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -324,6 +328,28 @@ public void setAuthorizationExceptionRetryInterval(Duration authorizationExcepti
 		this.authorizationExceptionRetryInterval = authorizationExceptionRetryInterval;
 	}
 
+	/**
+	 * The number of retries allowed when a
+	 * {@link org.apache.kafka.clients.consumer.RetriableCommitFailedException} is thrown
+	 * by the consumer.
+	 * @return the number of retries.
+	 * @since 2.3.9
+	 */
+	public int getCommitRetries() {
+		return this.commitRetries;
+	}
+
+	/**
+	 * Set number of retries allowed when a
+	 * {@link org.apache.kafka.clients.consumer.RetriableCommitFailedException} is thrown
+	 * by the consumer. Default 3 (4 attempts total).
+	 * @param commitRetries the commitRetries.
+	 * @since 2.3.9
+	 */
+	public void setCommitRetries(int commitRetries) {
+		this.commitRetries = commitRetries;
+	}
+
 	@Override
 	public String toString() {
 		return "ConsumerProperties ["

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

@@ -49,6 +49,7 @@
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.clients.consumer.RetriableCommitFailedException;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
@@ -1201,6 +1202,10 @@ private void wrapUp() {
 		 * @param e the exception.
 		 */
 		protected void handleConsumerException(Exception e) {
+			if (e instanceof RetriableCommitFailedException) {
+				this.logger.error(e, "Commit retries exhausted");
+				return;
+			}
 			try {
 				if (!this.isBatchListener && this.errorHandler != null) {
 					this.errorHandler.handle(e, Collections.emptyList(), this.consumer,
@@ -1280,13 +1285,25 @@ private void ackImmediate(ConsumerRecord<K, V> record) {
 				this.producer.sendOffsetsToTransaction(commits, this.consumerGroupId);
 			}
 			else if (this.syncCommits) {
-				this.consumer.commitSync(commits, this.syncCommitTimeout);
+				commitSync(commits);
 			}
 			else {
-				this.consumer.commitAsync(commits, this.commitCallback);
+				commitAsync(commits, 0);
 			}
 		}
 
+		private void commitAsync(Map<TopicPartition, OffsetAndMetadata> commits, int retries) {
+			this.consumer.commitAsync(commits, (offsetsAttempted, exception) -> {
+				if (exception instanceof RetriableCommitFailedException
+						&& retries < this.containerProperties.getCommitRetries()) {
+					commitAsync(commits, retries + 1);
+				}
+				else {
+					this.commitCallback.onComplete(offsetsAttempted, exception);
+				}
+			});
+		}
+
 		private void invokeListener(final ConsumerRecords<K, V> records) {
 			if (this.isBatchListener) {
 				invokeBatchListener(records);
@@ -1848,10 +1865,10 @@ public void ackCurrent(final ConsumerRecord<K, V> record,
 				if (producer == null) {
 					this.commitLogger.log(() -> "Committing: " + offsetsToCommit);
 					if (this.syncCommits) {
-						this.consumer.commitSync(offsetsToCommit, this.syncCommitTimeout);
+						commitSync(offsetsToCommit);
 					}
 					else {
-						this.consumer.commitAsync(offsetsToCommit, this.commitCallback);
+						commitAsync(offsetsToCommit, 0);
 					}
 				}
 				else {
@@ -2071,10 +2088,10 @@ private void commitIfNecessary() {
 				this.commitLogger.log(() -> "Committing: " + commits);
 				try {
 					if (this.syncCommits) {
-						this.consumer.commitSync(commits, this.syncCommitTimeout);
+						commitSync(commits);
 					}
 					else {
-						this.consumer.commitAsync(commits, this.commitCallback);
+						commitAsync(commits, 0);
 					}
 				}
 				catch (@SuppressWarnings(UNUSED) WakeupException e) {
@@ -2084,6 +2101,22 @@ private void commitIfNecessary() {
 			}
 		}
 
+		private void commitSync(Map<TopicPartition, OffsetAndMetadata> commits) {
+			doCommitSync(commits, 0);
+		}
+
+		private void doCommitSync(Map<TopicPartition, OffsetAndMetadata> commits, int retries) {
+			try {
+				this.consumer.commitSync(commits, this.syncCommitTimeout);
+			}
+			catch (RetriableCommitFailedException e) {
+				if (retries >= this.containerProperties.getCommitRetries()) {
+					throw e;
+				}
+				doCommitSync(commits, retries + 1);
+			}
+		}
+
 		private Map<TopicPartition, OffsetAndMetadata> buildCommits() {
 			Map<TopicPartition, OffsetAndMetadata> commits = new HashMap<>();
 			for (Entry<String, Map<Integer, Long>> entry : this.offsets.entrySet()) {
@@ -2348,6 +2381,7 @@ private boolean collectAndCommitIfNecessary(Collection<TopicPartition> partition
 				return true;
 			}
 
+			@SuppressWarnings("unused")
 			private void commitCurrentOffsets(Map<TopicPartition, OffsetAndMetadata> offsetsToCommit) {
 				ListenerConsumer.this.commitLogger.log(() -> "Committing on assignment: " + offsetsToCommit);
 				if (ListenerConsumer.this.transactionTemplate != null
@@ -2387,12 +2421,16 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 				else {
 					ContainerProperties containerProps = KafkaMessageListenerContainer.this.getContainerProperties();
 					if (containerProps.isSyncCommits()) {
-						ListenerConsumer.this.consumer.commitSync(offsetsToCommit,
-								containerProps.getSyncCommitTimeout());
+						try {
+							ListenerConsumer.this.consumer.commitSync(offsetsToCommit,
+									containerProps.getSyncCommitTimeout());
+						}
+						catch (RetriableCommitFailedException e) {
+							// ignore since this is on assignment anyway
+						}
 					}
 					else {
-						ListenerConsumer.this.consumer.commitAsync(offsetsToCommit,
-								containerProps.getCommitCallback());
+						commitAsync(offsetsToCommit, 0);
 					}
 				}
 			}

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

@@ -66,6 +66,8 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+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.AuthorizationException;
@@ -2855,6 +2857,73 @@ public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
 		container.stop();
 	}
 
+	@Test
+	void testCommitSyncRetries() throws Exception {
+		testCommitRetriesGuts(true);
+	}
+
+	@Test
+	void testCommitAsyncRetries() throws Exception {
+		testCommitRetriesGuts(false);
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	private void testCommitRetriesGuts(boolean sync) 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);
+		Map<String, Object> cfProps = new HashMap<>();
+		cfProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 45000); // wins
+		given(cf.getConfigurationProperties()).willReturn(cfProps);
+		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
+		records.put(new TopicPartition("foo", 0), Arrays.asList(
+				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
+				new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
+		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		ConsumerRecords<Integer, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
+		AtomicBoolean first = new AtomicBoolean(true);
+		given(consumer.poll(any(Duration.class))).willAnswer(i -> {
+			Thread.sleep(50);
+			return first.getAndSet(false) ? consumerRecords : emptyRecords;
+		});
+		CountDownLatch latch = new CountDownLatch(4);
+		if (sync) {
+			willAnswer(i -> {
+				latch.countDown();
+				throw new RetriableCommitFailedException("");
+			}).given(consumer).commitSync(anyMap(), eq(Duration.ofSeconds(45)));
+		}
+		else {
+			willAnswer(i -> {
+				OffsetCommitCallback callback = i.getArgument(1);
+				callback.onComplete(i.getArgument(0), new RetriableCommitFailedException(""));
+				latch.countDown();
+				return null;
+			}).given(consumer).commitAsync(anyMap(), any());
+		}
+		TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
+				new TopicPartitionOffset("foo", 0) };
+		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+		containerProps.setSyncCommits(sync);
+		containerProps.setGroupId("grp");
+		containerProps.setClientId("clientId");
+		containerProps.setIdleEventInterval(100L);
+		containerProps.setMessageListener((MessageListener) r -> {
+		});
+		containerProps.setMissingTopicsFatal(false);
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		if (sync) {
+			verify(consumer, times(4)).commitSync(any(), any());
+		}
+		else {
+			verify(consumer, times(4)).commitAsync(any(), any());
+		}
+	}
+
 	private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
 		Consumer<?, ?> consumer =
 				KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class);