GH-1274: Fix STCEH Deprecated CTOR

Fixes #1274

The deprecated CTORs taking `maxFailures` did not properly set up the
`BackOff` when `maxFailures` was negative (infinite retries).

Author: garyrussell

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

@@ -68,10 +68,17 @@ protected FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Excep
 	 * @param maxFailures the max failures.
 	 */
 	FailedRecordProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, int maxFailures) {
-		this.failureTracker = new FailedRecordTracker(recoverer, new FixedBackOff(0L, maxFailures - 1), LOGGER);
+		this.failureTracker = new FailedRecordTracker(recoverer, maxFailuresToBackOff(maxFailures), LOGGER);
 		this.classifier = configureDefaultClassifier();
 	}
 
+	private static FixedBackOff maxFailuresToBackOff(int maxFailures) {
+		if (maxFailures < 0) {
+			return new FixedBackOff();
+		}
+		return new FixedBackOff(0L, maxFailures == 0 ? 0 : maxFailures - 1);
+	}
+
 	/**
 	 * Return the exception classifier.
 	 * @return the classifier.

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

@@ -35,6 +35,7 @@
 
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.serializer.DeserializationException;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 /**
  * @author Gary Russell
@@ -79,4 +80,17 @@ consumer, mock(MessageListenerContainer.class)))
 		inOrder.verifyNoMoreInteractions();
 	}
 
+	@SuppressWarnings("deprecation")
+	@Test
+	public void testDeprecatedCtor() {
+		SeekToCurrentErrorHandler handler = new SeekToCurrentErrorHandler(-1);
+		assertThat(KafkaTestUtils.getPropertyValue(handler, "failureTracker.backOff.maxAttempts"))
+				.isEqualTo(Long.MAX_VALUE);
+		handler = new SeekToCurrentErrorHandler(0);
+		assertThat(KafkaTestUtils.getPropertyValue(handler, "failureTracker.backOff.maxAttempts"))
+				.isEqualTo(0L);
+		handler = new SeekToCurrentErrorHandler(10);
+		assertThat(KafkaTestUtils.getPropertyValue(handler, "failureTracker.backOff.maxAttempts"))
+				.isEqualTo(9L);
+	}
 }