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.
@@ -348,6 +352,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

@@ -52,6 +52,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;
@@ -60,6 +61,7 @@
 import org.apache.kafka.common.errors.FencedInstanceIdException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.apache.kafka.common.errors.RebalanceInProgressException;
+import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.WakeupException;
 import org.apache.kafka.common.header.internals.RecordHeader;
 
@@ -1236,6 +1238,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,
@@ -1318,10 +1324,21 @@ else if (this.syncCommits) {
 				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 RetriableException && 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);
@@ -1903,7 +1920,7 @@ public void ackCurrent(final ConsumerRecord<K, V> record) {
 						commitSync(offsetsToCommit);
 					}
 					else {
-						this.consumer.commitAsync(offsetsToCommit, this.commitCallback);
+						commitAsync(offsetsToCommit, 0);
 					}
 				}
 				else {
@@ -2134,7 +2151,7 @@ private void commitIfNecessary() {
 						commitSync(commits);
 					}
 					else {
-						this.consumer.commitAsync(commits, this.commitCallback);
+						commitAsync(commits, 0);
 					}
 				}
 				catch (@SuppressWarnings(UNUSED) WakeupException e) {
@@ -2145,9 +2162,19 @@ 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);
+			}
 			catch (RebalanceInProgressException e) {
 				this.logger.debug(e, "Non-fatal commit failure");
 				this.commitsDuringRebalance.putAll(commits);
@@ -2459,13 +2486,12 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 							ListenerConsumer.this.consumer.commitSync(offsetsToCommit,
 									containerProps.getSyncCommitTimeout());
 						}
-						catch (RebalanceInProgressException e) {
+						catch (RetriableCommitFailedException | RebalanceInProgressException 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/DeadLetterPublishingRecovererTests.java

@@ -29,7 +29,7 @@
 import java.io.IOException;
 import java.io.ObjectOutputStream;
 import java.io.UncheckedIOException;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.Map;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -178,7 +178,7 @@ void tombstoneWithMultiTemplates() {
 		KafkaOperations<?, ?> template1 = mock(KafkaOperations.class);
 		given(template1.send(any(ProducerRecord.class))).willReturn(new SettableListenableFuture());
 		KafkaOperations<?, ?> template2 = mock(KafkaOperations.class);
-		Map<Class<?>, KafkaOperations<?, ?>> templates = new HashMap<>();
+		Map<Class<?>, KafkaOperations<?, ?>> templates = new LinkedHashMap<>();
 		templates.put(String.class, template1);
 		templates.put(Integer.class, template2);
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(templates);
@@ -193,7 +193,7 @@ void tombstoneWithMultiTemplatesExplicit() {
 		KafkaOperations<?, ?> template1 = mock(KafkaOperations.class);
 		KafkaOperations<?, ?> template2 = mock(KafkaOperations.class);
 		given(template2.send(any(ProducerRecord.class))).willReturn(new SettableListenableFuture());
-		Map<Class<?>, KafkaOperations<?, ?>> templates = new HashMap<>();
+		Map<Class<?>, KafkaOperations<?, ?>> templates = new LinkedHashMap<>();
 		templates.put(String.class, template1);
 		templates.put(Void.class, template2);
 		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(templates);

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;
@@ -3011,6 +3013,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);