GH-2404: Refactor FallbackBatchErrorHandler

Resolves #2404

- implement `CommonErrorHandler` instead of using an adapter
- populate its retry listeners and log level if needed
- remove listeners thread local in `ErrorHandlingUtils`, needed for previous implementation
- remove final use of `SeekToCurrentBatchErrorHandler` - used in the fallback
  if recovery fails or the thread is interrupted while sleeping
- remove `SeekToCurrentBatchErrorHandler` and its tests

* Add CTOR to private inner class.

Author: garyrussell

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

@@ -107,7 +107,7 @@ public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff
 	}
 
 	private static CommonErrorHandler createFallback(BackOff backOff, @Nullable ConsumerRecordRecoverer recoverer) {
-		return new ErrorHandlerAdapter(new FallbackBatchErrorHandler(backOff, recoverer));
+		return new FallbackBatchErrorHandler(backOff, recoverer);
 	}
 
 	/**

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

@@ -28,7 +28,6 @@
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.KafkaUtils;
-import org.springframework.lang.Nullable;
 import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.BackOffExecution;
 
@@ -41,28 +40,9 @@
  */
 public final class ErrorHandlingUtils {
 
-	private static final ThreadLocal<List<RetryListener>> retryListeners = new ThreadLocal<>();
-
 	private ErrorHandlingUtils() {
 	}
 
-	/**
-	 * Set the retry listeners.
-	 * @param listeners the listeners.
-	 * @since 2.8.10
-	 */
-	public static void setRetryListeners(List<RetryListener> listeners) {
-		retryListeners.set(listeners);
-	}
-
-	/**
-	 * Clear the retry listeners.
-	 * @since 2.8.10
-	 */
-	public static void clearRetryListeners() {
-		retryListeners.remove();
-	}
-
 	/**
 	 * Retry a complete batch by pausing the consumer and then, in a loop, poll the
 	 * consumer, wait for the next back off, then call the listener. When retries are
@@ -77,20 +57,20 @@ public static void clearRetryListeners() {
 	 * @param recoverer the recoverer.
 	 * @param logger the logger.
 	 * @param logLevel the log level.
+	 * @param retryListeners the retry listeners.
 	 */
 	public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container, Runnable invokeListener, BackOff backOff,
 			CommonErrorHandler seeker, BiConsumer<ConsumerRecords<?, ?>, Exception> recoverer, LogAccessor logger,
-			KafkaException.Level logLevel) {
+			KafkaException.Level logLevel, List<RetryListener> retryListeners) {
 
 		BackOffExecution execution = backOff.start();
 		long nextBackOff = execution.nextBackOff();
 		String failed = null;
 		Set<TopicPartition> assignment = consumer.assignment();
 		consumer.pause(assignment);
-		List<RetryListener> listeners = retryListeners.get();
 		int attempt = 1;
-		listen(listeners, records, thrownException, attempt++);
+		listen(retryListeners, records, thrownException, attempt++);
 		if (container instanceof KafkaMessageListenerContainer) {
 			((KafkaMessageListenerContainer<?, ?>) container).publishConsumerPausedEvent(assignment, "For batch retry");
 		}
@@ -113,7 +93,7 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 					return;
 				}
 				catch (Exception ex) {
-					listen(listeners, records, ex, attempt++);
+					listen(retryListeners, records, ex, attempt++);
 					if (failed == null) {
 						failed = recordsToString(records);
 					}
@@ -124,15 +104,11 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 			}
 			try {
 				recoverer.accept(records, thrownException);
-				if (listeners != null) {
-					listeners.forEach(listener -> listener.recovered(records, thrownException));
-				}
+				retryListeners.forEach(listener -> listener.recovered(records, thrownException));
 			}
 			catch (Exception ex) {
 				logger.error(ex, () -> "Recoverer threw an exception; re-seeking batch");
-				if (listeners != null) {
-					listeners.forEach(listener -> listener.recoveryFailed(records, thrownException, ex));
-				}
+				retryListeners.forEach(listener -> listener.recoveryFailed(records, thrownException, ex));
 				seeker.handleBatch(thrownException, records, consumer, container, () -> { });
 			}
 		}
@@ -145,12 +121,10 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 		}
 	}
 
-	private static void listen(@Nullable List<RetryListener> listeners, ConsumerRecords<?, ?> records,
+	private static void listen(List<RetryListener> listeners, ConsumerRecords<?, ?> records,
 			Exception thrownException, int attempt) {
 
-		if (listeners != null) {
-			listeners.forEach(listener -> listener.failedDelivery(records, thrownException, attempt));
-		}
+		listeners.forEach(listener -> listener.failedDelivery(records, thrownException, attempt));
 	}
 
 	/**

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

@@ -35,6 +35,7 @@
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.KafkaException.Level;
 import org.springframework.lang.Nullable;
 import org.springframework.util.backoff.BackOff;
 
@@ -84,6 +85,22 @@ public FailedBatchProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception
 		this.fallbackBatchHandler = fallbackHandler;
 	}
 
+	@Override
+	public void setRetryListeners(RetryListener... listeners) {
+		super.setRetryListeners(listeners);
+		if (this.fallbackBatchHandler instanceof FallbackBatchErrorHandler handler) {
+			handler.setRetryListeners(listeners);
+		}
+	}
+
+	@Override
+	public void setLogLevel(Level logLevel) {
+		super.setLogLevel(logLevel);
+		if (this.fallbackBatchHandler instanceof KafkaExceptionLogLevelAware handler) {
+			handler.setLogLevel(logLevel);
+		}
+	}
+
 	/**
 	 * Return the fallback batch error handler.
 	 * @return the handler.
@@ -127,13 +144,7 @@ protected <K, V> ConsumerRecords<K, V> handle(Exception thrownException, Consume
 	private void fallback(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container, Runnable invokeListener) {
 
-		ErrorHandlingUtils.setRetryListeners(getRetryListeners());
-		try {
-			this.fallbackBatchHandler.handleBatch(thrownException, data, consumer, container, invokeListener);
-		}
-		finally {
-			ErrorHandlingUtils.clearRetryListeners();
-		}
+		this.fallbackBatchHandler.handleBatch(thrownException, data, consumer, container, invokeListener);
 	}
 
 	private int findIndex(ConsumerRecords<?, ?> data, ConsumerRecord<?, ?> record) {

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

@@ -16,16 +16,23 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.KafkaException;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.FixedBackOff;
 
@@ -41,23 +48,22 @@
  * @since 2.3.7
  *
  */
-@SuppressWarnings("deprecation")
-class FallbackBatchErrorHandler extends KafkaExceptionLogLevelAware
-		implements ListenerInvokingBatchErrorHandler {
+class FallbackBatchErrorHandler extends KafkaExceptionLogLevelAware implements CommonErrorHandler {
 
 	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
 
 	private final BackOff backOff;
 
 	private final BiConsumer<ConsumerRecords<?, ?>, Exception> recoverer;
 
-	@SuppressWarnings("deprecation")
-	private final CommonErrorHandler seeker = new ErrorHandlerAdapter(new SeekToCurrentBatchErrorHandler());
-
-	private boolean ackAfterHandle = true;
+	private final CommonErrorHandler seeker = new SeekAfterRecoverFailsOrInterrupted();
 
 	private final ThreadLocal<Boolean> retrying = ThreadLocal.withInitial(() -> false);
 
+	private final List<RetryListener> retryListeners = new ArrayList<>();
+
+	private boolean ackAfterHandle = true;
+
 	/**
 	 * Construct an instance with a default {@link FixedBackOff} (unlimited attempts with
 	 * a 5 second back off).
@@ -85,6 +91,18 @@ class FallbackBatchErrorHandler extends KafkaExceptionLogLevelAware
 		};
 	}
 
+	/**
+	 * Set one or more {@link RetryListener} to receive notifications of retries and
+	 * recovery.
+	 * @param listeners the listeners.
+	 * @since 3.0
+	 */
+	public void setRetryListeners(RetryListener... listeners) {
+		Assert.noNullElements(listeners, "'listeners' cannot have null elements");
+		this.retryListeners.clear();
+		this.retryListeners.addAll(Arrays.asList(listeners));
+	}
+
 	@Override
 	public boolean isAckAfterHandle() {
 		return this.ackAfterHandle;
@@ -96,7 +114,7 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 	}
 
 	@Override
-	public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> records,
+	public void handleBatch(Exception thrownException, @Nullable ConsumerRecords<?, ?> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
 
 		if (records == null || records.count() == 0) {
@@ -106,13 +124,14 @@ public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> re
 		this.retrying.set(true);
 		try {
 			ErrorHandlingUtils.retryBatch(thrownException, records, consumer, container, invokeListener, this.backOff,
-					this.seeker, this.recoverer, this.logger, getLogLevel());
+					this.seeker, this.recoverer, this.logger, getLogLevel(), this.retryListeners);
 		}
 		finally {
 			this.retrying.set(false);
 		}
 	}
 
+	@Override
 	public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions,
 			Runnable publishPause) {
 
@@ -122,4 +141,26 @@ public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartit
 		}
 	}
 
+	private final class SeekAfterRecoverFailsOrInterrupted implements CommonErrorHandler {
+
+		SeekAfterRecoverFailsOrInterrupted() {
+		}
+
+		@Override
+		public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+				MessageListenerContainer container, Runnable invokeListener) {
+
+			data.partitions()
+					.stream()
+					.collect(
+							Collectors.toMap(tp -> tp,
+									tp -> data.records(tp).get(0).offset(), (u, v) -> (long) v, LinkedHashMap::new))
+					.forEach(consumer::seek);
+
+			throw new KafkaException("Seek to current after exception", getLogLevel(), thrownException);
+
+		}
+
+	}
+
 }