GH-2076: Fix Async Commit Retries

#2076

Do not attempt to retry asynchronous commits.
- a later commit for the same topic/partition may have already succeeded
- many consecutive retryable exceptions can cause a stack overflow

**cherry-pick to 2.8.x, 2.7.x**

* Remove unused parameter; polish javadocs.
# Conflicts:
#	spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

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

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2021 the original author or authors.
+ * Copyright 2019-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -289,6 +289,7 @@ public OffsetCommitCallback getCommitCallback() {
 	 * @see #setSyncCommitTimeout(Duration)
 	 * @see #setCommitCallback(OffsetCommitCallback)
 	 * @see #setCommitLogLevel(org.springframework.kafka.support.LogIfLevelEnabled.Level)
+	 * @see #setCommitRetries(int)
 	 */
 	public void setSyncCommits(boolean syncCommits) {
 		this.syncCommits = syncCommits;
@@ -368,9 +369,10 @@ public void setAuthorizationExceptionRetryInterval(Duration authorizationExcepti
 	/**
 	 * The number of retries allowed when a
 	 * {@link org.apache.kafka.clients.consumer.RetriableCommitFailedException} is thrown
-	 * by the consumer.
+	 * by the consumer when using {@link #setSyncCommits(boolean)} set to true.
 	 * @return the number of retries.
 	 * @since 2.3.9
+	 * @see #setSyncCommits(boolean)
 	 */
 	public int getCommitRetries() {
 		return this.commitRetries;
@@ -379,9 +381,11 @@ public int getCommitRetries() {
 	/**
 	 * Set number of retries allowed when a
 	 * {@link org.apache.kafka.clients.consumer.RetriableCommitFailedException} is thrown
-	 * by the consumer. Default 3 (4 attempts total).
+	 * by the consumer when using {@link #setSyncCommits(boolean)} set to true. Default 3
+	 * (4 attempts total).
 	 * @param commitRetries the commitRetries.
 	 * @since 2.3.9
+	 * @see #setSyncCommits(boolean)
 	 */
 	public void setCommitRetries(int commitRetries) {
 		this.commitRetries = commitRetries;

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

@@ -1368,7 +1368,7 @@ private void fixTxOffsetsIfNeeded() {
 								commitSync(toFix);
 							}
 							else {
-								commitAsync(toFix, 0);
+								commitAsync(toFix);
 							}
 						}
 						else {
@@ -1682,18 +1682,14 @@ else if (this.syncCommits) {
 				commitSync(commits);
 			}
 			else {
-				commitAsync(commits, 0);
+				commitAsync(commits);
 			}
 		}
 
-		private void commitAsync(Map<TopicPartition, OffsetAndMetadata> commits, int retries) {
+		private void commitAsync(Map<TopicPartition, OffsetAndMetadata> commits) {
 			this.consumer.commitAsync(commits, (offsetsAttempted, exception) -> {
-				if (exception instanceof RetriableCommitFailedException
-						&& retries < this.containerProperties.getCommitRetries()) {
-					commitAsync(commits, retries + 1);
-				}
-				else {
-					this.commitCallback.onComplete(offsetsAttempted, exception);
+				this.commitCallback.onComplete(offsetsAttempted, exception);
+				if (exception == null) {
 					if (this.fixTxOffsets) {
 						this.lastCommits.putAll(commits);
 					}
@@ -2451,7 +2447,7 @@ public void ackCurrent(final ConsumerRecord<K, V> record) {
 						commitSync(offsetsToCommit);
 					}
 					else {
-						commitAsync(offsetsToCommit, 0);
+						commitAsync(offsetsToCommit);
 					}
 				}
 				else {
@@ -2706,7 +2702,7 @@ private void commitIfNecessary() {
 						commitSync(commits);
 					}
 					else {
-						commitAsync(commits, 0);
+						commitAsync(commits);
 					}
 				}
 				catch (@SuppressWarnings(UNUSED) WakeupException e) {
@@ -3051,7 +3047,7 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
 						}
 					}
 					else {
-						commitAsync(offsetsToCommit, 0);
+						commitAsync(offsetsToCommit);
 					}
 				}
 			}

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

@@ -70,7 +70,6 @@
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 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.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
@@ -3124,17 +3123,8 @@ public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
 	}
 
 	@Test
-	void testCommitSyncRetries() throws Exception {
-		testCommitRetriesGuts(true);
-	}
-
-	@Test
-	void testCommitAsyncRetries() throws Exception {
-		testCommitRetriesGuts(false);
-	}
-
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	private void testCommitRetriesGuts(boolean sync) throws Exception {
+	void testCommitSyncRetries() 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);
@@ -3153,24 +3143,14 @@ private void testCommitRetriesGuts(boolean sync) throws Exception {
 			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);
+		willAnswer(i -> {
+			latch.countDown();
+			throw new RetriableCommitFailedException("");
+		}).given(consumer).commitSync(anyMap(), eq(Duration.ofSeconds(45)));
+		containerProps.setSyncCommits(true);
 		containerProps.setGroupId("grp");
 		containerProps.setClientId("clientId");
 		containerProps.setIdleEventInterval(100L);
@@ -3182,12 +3162,7 @@ private void testCommitRetriesGuts(boolean sync) throws Exception {
 		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());
-		}
+		verify(consumer, times(4)).commitSync(any(), any());
 	}
 
 	@Test