GH-1804: Add ExponentialBackOffWithMaxRetries

Resolves #1804

Simply a convenience to automatically calculate the max elapsed time.

Author: garyrussell

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

@@ -5587,6 +5587,29 @@ Starting with version 2.7, the recoverer checks that the partition selected by t
 If the partition is not present, the partition in the `ProducerRecord` is set to `null`, allowing the `KafkaProducer` to select the partition.
 You can disable this check by setting the `verifyPartition` property to `false`.
 
+[[exp-backoff]]
+===== `ExponentialBackOffWithMaxRetries` Implementation
+
+Spring Framework provides a number of `BackOff` implementations.
+By default, the `ExponentialBackOff` will retry indefinitely; to give up after some number of retry attempts requires calculating the `maxElapsedTime`.
+Since version 2.7.3, Spring for Apache Kafka provides the `ExponentialBackOffWithMaxRetries` which is a subclass that receives the `maxRetries` property and automatically calculates the `maxElapsedTime`, which is a little more convenient.
+
+====
+[source, java]
+----
+@Bean
+SeekToCurrentErrorHandler handler() {
+    ExponentialBackOffWithMaxRetries bo = new ExponentialBackOffWithMaxRetries(6);
+    bo.setInitialInterval(1_000L);
+    bo.setMultiplier(2.0);
+    bo.setMaxInterval(10_000L);
+    return new SeekToCurrentErrorHandler(myRecoverer, bo);
+}
+----
+====
+
+This will retry after `1, 2, 4, 8, 10, 10` seconds, before calling the recoverer.
+
 [[kerberos]]
 ==== JAAS and Kerberos
 

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

@@ -99,3 +99,9 @@ See <<messaging-message-conversion>> for more information.
 ==== Sequencing `@KafkaListener` s
 
 See <<container-sequencing>> for more information.
+
+[[x27-exp-backoff]]
+==== `ExponentialBackOffWithMaxRetries`
+
+A new `BackOff` implementation is provided, making it more convenient to configure the max retries.
+See <<exp-backoff>> for more information.

spring-kafka/src/main/java/org/springframework/kafka/support/ExponentialBackOffWithMaxRetries.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support;
+
+import org.springframework.util.backoff.ExponentialBackOff;
+
+/**
+ * Subclass of {@link ExponentialBackOff} that allows the specification of the maximum
+ * number of retries rather than the maximum elapsed time.
+ *
+ * @author Gary Russell
+ * @since 2.7.3
+ *
+ */
+public class ExponentialBackOffWithMaxRetries extends ExponentialBackOff {
+
+	private final int maxRetries;
+
+	/**
+	 * Construct an instance that will calculate the {@link #setMaxElapsedTime(long)} from
+	 * the maxRetries.
+	 * @param maxRetries the max retries.
+	 */
+	public ExponentialBackOffWithMaxRetries(int maxRetries) {
+		this.maxRetries = maxRetries;
+		calculateMaxElapsed();
+	}
+
+	/**
+	 * Get the max retries.
+	 * @return the max retries.
+	 */
+	public int getMaxRetries() {
+		return this.maxRetries;
+	}
+
+	@Override
+	public void setInitialInterval(long initialInterval) {
+		super.setInitialInterval(initialInterval);
+		calculateMaxElapsed();
+	}
+
+	@Override
+	public void setMultiplier(double multiplier) {
+		super.setMultiplier(multiplier);
+		calculateMaxElapsed();
+	}
+
+	@Override
+	public void setMaxInterval(long maxInterval) {
+		super.setMaxInterval(maxInterval);
+		calculateMaxElapsed();
+	}
+
+	@Override
+	public void setMaxElapsedTime(long maxElapsedTime) {
+		throw new IllegalStateException("'maxElapsedTime' is calculated from the 'maxRetries' property");
+	}
+
+	private void calculateMaxElapsed() {
+		long maxInterval = getMaxInterval();
+		long maxElapsed = Math.min(getInitialInterval(), maxInterval);
+		long current = maxElapsed;
+		for (int i = 1; i < this.maxRetries; i++) {
+			long next = Math.min((long) (current * getMultiplier()), maxInterval);
+			current = next;
+			maxElapsed += current;
+		}
+		super.setMaxElapsedTime(maxElapsed);
+	}
+
+}

spring-kafka/src/test/java/org/springframework/kafka/support/ExponentialBackOffWithMaxRetriesTests.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.IntStream;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.util.backoff.BackOffExecution;
+
+/**
+ * @author Gary Russell
+ * @since 2.7.3
+ *
+ */
+public class ExponentialBackOffWithMaxRetriesTests {
+
+	@Test
+	void calcAll() {
+		ExponentialBackOffWithMaxRetries bo = new ExponentialBackOffWithMaxRetries(10);
+		bo.setInitialInterval(1_000L);
+		bo.setMultiplier(2.0);
+		bo.setMaxInterval(10_000L);
+		assertThatIllegalStateException().isThrownBy(() -> bo.setMaxElapsedTime(42L));
+		assertThat(bo.getMaxRetries()).isEqualTo(10);
+		List<Long> delays = new ArrayList<>();
+		BackOffExecution boEx = bo.start();
+		IntStream.range(0, 11).forEach(i -> delays.add(boEx.nextBackOff()));
+		assertThat(delays).containsExactly(1_000L, 2_000L, 4_000L, 8_000L, 10_000L, 10_000L, 10_000L, 10_000L, 10_000L,
+				10_000L, -1L);
+	}
+
+	@Test
+	void calcMaxLessThanInitial() {
+		ExponentialBackOffWithMaxRetries bo = new ExponentialBackOffWithMaxRetries(3);
+		bo.setInitialInterval(10_000L);
+		bo.setMultiplier(2.0);
+		bo.setMaxInterval(5_000L);
+		List<Long> delays = new ArrayList<>();
+		BackOffExecution boEx = bo.start();
+		IntStream.range(0, 4).forEach(i -> delays.add(boEx.nextBackOff()));
+		assertThat(delays).containsExactly(5_000L, 5_000L, 5_000L, -1L);
+	}
+
+	@Test
+	void calcDefault() {
+		ExponentialBackOffWithMaxRetries bo = new ExponentialBackOffWithMaxRetries(10);
+		List<Long> delays = new ArrayList<>();
+		BackOffExecution boEx = bo.start();
+		IntStream.range(0, 11).forEach(i -> delays.add(boEx.nextBackOff()));
+		assertThat(delays).containsExactly(2_000L, 3_000L, 4_500L, 6_750L, 10_125L, 15_187L, 22_780L, 30_000L, 30_000L,
+				30_000L, -1L);
+	}
+
+}