GH-2459: FallbackBatchErrorHandler Retryable Ex

Resolves #2459

The `FallbackBatchErrorHandler` was not an `ExceptionClassifier`.
The default error handler should propagate exception classifications.

**2.9.x - I will back port**

Author: garyrussell

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

@@ -19,13 +19,16 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Stream;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.kafka.support.TopicPartitionOffset;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -35,7 +38,7 @@
  * @since 2.7.4
  *
  */
-class ErrorHandlerAdapter implements CommonErrorHandler {
+class ErrorHandlerAdapter extends ExceptionClassifier implements CommonErrorHandler {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	private static final ConsumerRecords EMPTY_BATCH = new ConsumerRecords(Collections.emptyMap());
@@ -171,5 +174,30 @@ public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartit
 		}
 	}
 
+	@Override
+	protected void notRetryable(Stream<Class<? extends Exception>> notRetryable) {
+		if (this.batchErrorHandler instanceof ExceptionClassifier) {
+			notRetryable.forEach(ex -> ((ExceptionClassifier) this.batchErrorHandler).addNotRetryableExceptions(ex));
+		}
+	}
+
+	@Override
+	public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifications, boolean defaultValue) {
+		super.setClassifications(classifications, defaultValue);
+		if (this.batchErrorHandler instanceof ExceptionClassifier) {
+			((ExceptionClassifier) this.batchErrorHandler).setClassifications(classifications, defaultValue);
+		}
+	}
+
+	@Override
+	@Nullable
+	public Boolean removeClassification(Class<? extends Exception> exceptionType) {
+		Boolean removed = super.removeClassification(exceptionType);
+		if (this.batchErrorHandler instanceof ExceptionClassifier) {
+			((ExceptionClassifier) this.batchErrorHandler).removeClassification(exceptionType);
+		}
+		return removed;
+	}
+
 }
 

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

@@ -26,6 +26,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.lang.Nullable;
@@ -77,18 +78,49 @@ public static void clearRetryListeners() {
 	 * @param recoverer the recoverer.
 	 * @param logger the logger.
 	 * @param logLevel the log level.
+	 * @deprecated in favor of
+	 * {@link #retryBatch(Exception, ConsumerRecords, Consumer, MessageListenerContainer, Runnable, BackOff, CommonErrorHandler, BiConsumer, LogAccessor, org.springframework.kafka.KafkaException.Level, List, BinaryExceptionClassifier)}.
 	 */
+	@Deprecated
 	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) {
 
+		retryBatch(thrownException, records, consumer, container, invokeListener, backOff, seeker, null, logger,
+				logLevel, null, new BinaryExceptionClassifier(true));
+	}
+
+	/**
+	 * 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
+	 * exhausted, call the recoverer with the {@link ConsumerRecords}.
+	 * @param thrownException the exception.
+	 * @param records the records.
+	 * @param consumer the consumer.
+	 * @param container the container.
+	 * @param invokeListener the {@link Runnable} to run (call the listener).
+	 * @param backOff the backOff.
+	 * @param seeker the common error handler that re-seeks the entire batch.
+	 * @param recoverer the recoverer.
+	 * @param logger the logger.
+	 * @param logLevel the log level.
+	 * @param retryListenersArg the retry listeners.
+	 * @param classifier the exception classifier.
+	 * @since 2.8.11
+	 */
+	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, @Nullable List<RetryListener> retryListenersArg,
+			BinaryExceptionClassifier classifier) {
+
 		BackOffExecution execution = backOff.start();
 		long nextBackOff = execution.nextBackOff();
 		String failed = null;
 		Set<TopicPartition> assignment = consumer.assignment();
 		consumer.pause(assignment);
-		List<RetryListener> listeners = retryListeners.get();
+		List<RetryListener> listeners = retryListenersArg != null ? retryListenersArg : retryListeners.get();
 		int attempt = 1;
 		listen(listeners, records, thrownException, attempt++);
 		ConsumerRecord<?, ?> first = records.iterator().next();
@@ -98,7 +130,8 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 					.publishConsumerPausedEvent(assignment, "For batch retry");
 		}
 		try {
-			while (nextBackOff != BackOffExecution.STOP) {
+			Boolean retryable = classifier.classify(unwrapIfNeeded(thrownException));
+			while (Boolean.TRUE.equals(retryable) && nextBackOff != BackOffExecution.STOP) {
 				consumer.poll(Duration.ZERO);
 				try {
 					ListenerUtils.stoppableSleep(container, nextBackOff);
@@ -171,4 +204,22 @@ public static String recordsToString(ConsumerRecords<?, ?> records) {
 		return sb.toString();
 	}
 
+	/**
+	 * Remove a {@link TimestampedException}, if present.
+	 * Remove a {@link ListenerExecutionFailedException}, if present.
+	 * @param exception the exception.
+	 * @return the unwrapped cause or cause of cause.
+	 * @since 2.8.11
+	 */
+	public static Exception unwrapIfNeeded(Exception exception) {
+		Exception theEx = exception;
+		if (theEx instanceof TimestampedException && theEx.getCause() instanceof Exception) {
+			theEx = (Exception) theEx.getCause();
+		}
+		if (theEx instanceof ListenerExecutionFailedException && theEx.getCause() instanceof Exception) {
+			theEx = (Exception) theEx.getCause();
+		}
+		return theEx;
+	}
+
 }

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

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.kafka.support.converter.ConversionException;
@@ -130,6 +131,16 @@ public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifi
 	@SuppressWarnings("varargs")
 	public final void addNotRetryableExceptions(Class<? extends Exception>... exceptionTypes) {
 		add(false, exceptionTypes);
+		notRetryable(Arrays.stream(exceptionTypes));
+	}
+
+	/**
+	 * Subclasses can override this to receive notification of configuration of not
+	 * retryable exceptions.
+	 * @param notRetryable the not retryable exceptions.
+	 * @since 2.9.3
+	 */
+	protected void notRetryable(Stream<Class<? extends Exception>> notRetryable) {
 	}
 
 	/**

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

@@ -26,6 +26,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.function.BiConsumer;
+import java.util.stream.Stream;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -35,6 +36,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 +86,39 @@ public FailedBatchProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception
 		this.fallbackBatchHandler = fallbackHandler;
 	}
 
+	@Override
+	public void setLogLevel(Level logLevel) {
+		super.setLogLevel(logLevel);
+		if (this.fallbackBatchHandler instanceof KafkaExceptionLogLevelAware) {
+			((KafkaExceptionLogLevelAware) this.fallbackBatchHandler).setLogLevel(logLevel);
+		}
+	}
+
+	@Override
+	protected void notRetryable(Stream<Class<? extends Exception>> notRetryable) {
+		if (this.fallbackBatchHandler instanceof ExceptionClassifier) {
+			notRetryable.forEach(ex -> ((ExceptionClassifier) this.fallbackBatchHandler).addNotRetryableExceptions(ex));
+		}
+	}
+
+	@Override
+	public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifications, boolean defaultValue) {
+		super.setClassifications(classifications, defaultValue);
+		if (this.fallbackBatchHandler instanceof ExceptionClassifier) {
+			((ExceptionClassifier) this.fallbackBatchHandler).setClassifications(classifications, defaultValue);
+		}
+	}
+
+	@Override
+	@Nullable
+	public Boolean removeClassification(Class<? extends Exception> exceptionType) {
+		Boolean removed = super.removeClassification(exceptionType);
+		if (this.fallbackBatchHandler instanceof ExceptionClassifier) {
+			((ExceptionClassifier) this.fallbackBatchHandler).removeClassification(exceptionType);
+		}
+		return removed;
+	}
+
 	/**
 	 * Return the fallback batch error handler.
 	 * @return the handler.

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

@@ -41,7 +41,7 @@
  * @since 2.3.7
  *
  */
-class FallbackBatchErrorHandler extends KafkaExceptionLogLevelAware
+class FallbackBatchErrorHandler extends ExceptionClassifier
 		implements ListenerInvokingBatchErrorHandler {
 
 	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
@@ -105,7 +105,7 @@ 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(), null, getClassifier());
 		}
 		finally {
 			this.retrying.set(false);

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

@@ -24,6 +24,7 @@
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -234,6 +235,33 @@ void fallbackListener() {
 		verify(retryListener).recovered(any(ConsumerRecords.class), any());
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void notRetryable() {
+		Consumer mockConsumer = mock(Consumer.class);
+		ConsumerRecordRecoverer recoverer = mock(ConsumerRecordRecoverer.class);
+		DefaultErrorHandler beh = new DefaultErrorHandler(recoverer, new FixedBackOff(0, 2));
+		beh.addNotRetryableExceptions(IllegalStateException.class);
+		RetryListener retryListener = mock(RetryListener.class);
+		beh.setRetryListeners(retryListener);
+		TopicPartition tp = new TopicPartition("foo", 0);
+		ConsumerRecords<?, ?> records = new ConsumerRecords(Collections.singletonMap(tp,
+				List.of(new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo",
+								new RecordHeaders(), Optional.empty()),
+						new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, null, "foo",
+								new RecordHeaders(), Optional.empty()))));
+		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		given(container.isRunning()).willReturn(true);
+		beh.handleBatch(new ListenerExecutionFailedException("test", new IllegalStateException()),
+				records, mockConsumer, container, () -> {
+				});
+		verify(retryListener).failedDelivery(any(ConsumerRecords.class), any(), eq(1));
+		// no retries
+		verify(retryListener, never()).failedDelivery(any(ConsumerRecords.class), any(), eq(2));
+		verify(recoverer, times(2)).accept(any(), any()); // each record in batch
+		verify(retryListener).recovered(any(ConsumerRecords.class), any());
+	}
+
 	@Configuration
 	@EnableKafka
 	public static class Config {