GH-2099: Deprecate RetryingBatchErrorHandler

Resolves https://github.com/spring-projects/spring-kafka/issues/2099

Similar to other legacy error handlers, deprecate this one because
its functionality is replaced by the `DefaultErrorHandler`.

Eventually move logic to `FallbackBatchErrorHandler` which is package
private.

**cherry-pick to 2.8.x**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-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.
@@ -36,11 +36,11 @@ import org.springframework.util.backoff.BackOff;
  * the partitions for the remaining records will be repositioned and/or the failed record
  * can be recovered and skipped. If some other exception is thrown, or a valid record is
  * not provided in the exception, error handling is delegated to a
- * {@link RetryingBatchErrorHandler} with this handler's {@link BackOff}. If the record is
+ * {@link FallbackBatchErrorHandler} with this handler's {@link BackOff}. If the record is
  * recovered, its offset is committed. This is a replacement for the legacy
  * {@link SeekToCurrentErrorHandler} and {@link SeekToCurrentBatchErrorHandler} (but the
  * fallback now can send the messages to a recoverer after retries are completed instead
- * of retring indefinitely).
+ * of retrying indefinitely).
  *
  * @author Gary Russell
  *
@@ -90,7 +90,7 @@ public class DefaultErrorHandler extends FailedBatchProcessor implements CommonE
 	}
 
 	private static CommonErrorHandler createFallback(BackOff backOff, @Nullable ConsumerRecordRecoverer recoverer) {
-		return new ErrorHandlerAdapter(new RetryingBatchErrorHandler(backOff, recoverer));
+		return new ErrorHandlerAdapter(new FallbackBatchErrorHandler(backOff, recoverer));
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-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.
@@ -54,7 +54,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor {
 
 	private static final LoggingCommitCallback LOGGING_COMMIT_CALLBACK = new LoggingCommitCallback();
 
-	private final CommonErrorHandler fallbackHandler;
+	private final CommonErrorHandler fallbackBatchHandler;
 
 	/**
 	 * Construct an instance with the provided properties.
@@ -66,7 +66,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor {
 			CommonErrorHandler fallbackHandler) {
 
 		super(recoverer, backOff);
-		this.fallbackHandler = fallbackHandler;
+		this.fallbackBatchHandler = fallbackHandler;
 	}
 
 	protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
@@ -75,7 +75,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor {
 		BatchListenerFailedException batchListenerFailedException = getBatchListenerFailedException(thrownException);
 		if (batchListenerFailedException == null) {
 			this.logger.debug(thrownException, "Expected a BatchListenerFailedException; re-seeking batch");
-			this.fallbackHandler.handleBatch(thrownException, data, consumer, container, invokeListener);
+			this.fallbackBatchHandler.handleBatch(thrownException, data, consumer, container, invokeListener);
 		}
 		else {
 			ConsumerRecord<?, ?> record = batchListenerFailedException.getRecord();
@@ -84,7 +84,7 @@ public abstract class FailedBatchProcessor extends FailedRecordProcessor {
 				this.logger.warn(batchListenerFailedException, () ->
 						String.format("Record not found in batch: %s-%d@%d; re-seeking batch",
 								record.topic(), record.partition(), record.offset()));
-				this.fallbackHandler.handleBatch(thrownException, data, consumer, container, invokeListener);
+				this.fallbackBatchHandler.handleBatch(thrownException, data, consumer, container, invokeListener);
 			}
 			else {
 				seekOrRecover(thrownException, data, consumer, container, index);

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 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.
+ * 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.listener;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.backoff.BackOff;
+import org.springframework.util.backoff.FixedBackOff;
+
+/**
+ * A batch error handler used by the default error handler when the listener does
+ * not throw a {@link BatchListenerFailedException}.
+ *
+ * @author Gary Russell
+ * @since 2.8.3
+ *
+ */
+@SuppressWarnings("deprecation")
+class FallbackBatchErrorHandler extends RetryingBatchErrorHandler {
+
+	/**
+	 * Construct an instance with a default {@link FixedBackOff} (unlimited attempts with
+	 * a 5 second back off).
+	 */
+	FallbackBatchErrorHandler() {
+		super();
+	}
+
+	/**
+	 * Construct an instance with the provided {@link BackOff} and
+	 * {@link ConsumerRecordRecoverer}. If the recoverer is {@code null}, the discarded
+	 * records (topic-partition{@literal @}offset) will be logged.
+	 * @param backOff the back off.
+	 * @param recoverer the recoverer.
+	 */
+	FallbackBatchErrorHandler(BackOff backOff, @Nullable ConsumerRecordRecoverer recoverer) {
+		super(backOff, recoverer);
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-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.
@@ -37,12 +37,14 @@ import org.springframework.util.backoff.FixedBackOff;
  *
  * @author Gary Russell
  * @since 2.3.7
+ * @deprecated in favor of {@link DefaultErrorHandler}.
  *
  */
+@Deprecated
 public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware
 		implements ListenerInvokingBatchErrorHandler {
 
-	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(RetryingBatchErrorHandler.class));
+	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
 
 	private final BackOff backOff;
 
@@ -72,7 +74,7 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware
 		this.backOff = backOff;
 		this.recoverer = (crs, ex) -> {
 			if (recoverer == null) {
-				LOGGER.error(ex, () -> "Records discarded: " + ErrorHandlingUtils.recordsToString(crs));
+				this.logger.error(ex, () -> "Records discarded: " + ErrorHandlingUtils.recordsToString(crs));
 			}
 			else {
 				crs.spliterator().forEachRemaining(rec -> recoverer.accept(rec, ex));
@@ -95,11 +97,11 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware
 			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
 
 		if (records == null || records.count() == 0) {
-			LOGGER.error(thrownException, "Called with no records; consumer exception");
+			this.logger.error(thrownException, "Called with no records; consumer exception");
 			return;
 		}
 		ErrorHandlingUtils.retryBatch(thrownException, records, consumer, container, invokeListener, this.backOff,
-				this.seeker, this.recoverer, LOGGER, getLogLevel());
+				this.seeker, this.recoverer, this.logger, getLogLevel());
 	}
 
 }

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

@@ -56,11 +56,11 @@ import org.springframework.util.backoff.FixedBackOff;
  *
  */
 @EmbeddedKafka(topics = {
-		RetryingBatchErrorHandlerIntegrationTests.topic1,
-		RetryingBatchErrorHandlerIntegrationTests.topic1DLT,
-		RetryingBatchErrorHandlerIntegrationTests.topic2,
-		RetryingBatchErrorHandlerIntegrationTests.topic2DLT})
-public class RetryingBatchErrorHandlerIntegrationTests {
+		FallbackBatchErrorHandlerIntegrationTests.topic1,
+		FallbackBatchErrorHandlerIntegrationTests.topic1DLT,
+		FallbackBatchErrorHandlerIntegrationTests.topic2,
+		FallbackBatchErrorHandlerIntegrationTests.topic2DLT})
+public class FallbackBatchErrorHandlerIntegrationTests {
 
 	public static final String topic1 = "retryTopic1";
 
@@ -115,7 +115,7 @@ public class RetryingBatchErrorHandlerIntegrationTests {
 			}
 
 		};
-		RetryingBatchErrorHandler errorHandler = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3), recoverer);
+		FallbackBatchErrorHandler errorHandler = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3), recoverer);
 		container.setBatchErrorHandler(errorHandler);
 		final CountDownLatch stopLatch = new CountDownLatch(1);
 		container.setApplicationEventPublisher(e -> {
@@ -186,7 +186,7 @@ public class RetryingBatchErrorHandlerIntegrationTests {
 			}
 
 		};
-		RetryingBatchErrorHandler errorHandler = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3), recoverer);
+		FallbackBatchErrorHandler errorHandler = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3), recoverer);
 		container.setBatchErrorHandler(errorHandler);
 		final CountDownLatch stopLatch = new CountDownLatch(1);
 		container.setApplicationEventPublisher(e -> {
@@ -226,7 +226,7 @@ public class RetryingBatchErrorHandlerIntegrationTests {
 		KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
 				containerProps);
 		CountDownLatch called = new CountDownLatch(1);
-		container.setBatchErrorHandler(new RetryingBatchErrorHandler() {
+		container.setBatchErrorHandler(new FallbackBatchErrorHandler() {
 
 			@Override
 			public void handle(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-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.
@@ -43,7 +43,7 @@ import org.springframework.util.backoff.FixedBackOff;
  * @since 2.3.7
  *
  */
-public class RetryingBatchErrorHandlerTests {
+public class FallbackBatchErrorHandlerTests {
 
 	private int invoked;
 
@@ -51,7 +51,7 @@ public class RetryingBatchErrorHandlerTests {
 	void recover() {
 		this.invoked = 0;
 		List<ConsumerRecord<?, ?>> recovered = new ArrayList<>();
-		RetryingBatchErrorHandler eh = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) ->  {
+		FallbackBatchErrorHandler eh = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) ->  {
 			recovered.add(cr);
 		});
 		Map<TopicPartition, List<ConsumerRecord<Object, Object>>> map = new HashMap<>();
@@ -79,7 +79,7 @@ public class RetryingBatchErrorHandlerTests {
 	void successOnRetry() {
 		this.invoked = 0;
 		List<ConsumerRecord<?, ?>> recovered = new ArrayList<>();
-		RetryingBatchErrorHandler eh = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) ->  {
+		FallbackBatchErrorHandler eh = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) ->  {
 			recovered.add(cr);
 		});
 		Map<TopicPartition, List<ConsumerRecord<Object, Object>>> map = new HashMap<>();
@@ -104,7 +104,7 @@ public class RetryingBatchErrorHandlerTests {
 	void recoveryFails() {
 		this.invoked = 0;
 		List<ConsumerRecord<?, ?>> recovered = new ArrayList<>();
-		RetryingBatchErrorHandler eh = new RetryingBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) ->  {
+		FallbackBatchErrorHandler eh = new FallbackBatchErrorHandler(new FixedBackOff(0L, 3L), (cr, ex) ->  {
 			recovered.add(cr);
 			throw new RuntimeException("can't recover");
 		});