GH-2459: FallbackBatchErrorHandler Retryable Ex

Resolves #2459

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

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

Author: garyrussell

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.kafka.support.KafkaUtils;
@@ -59,11 +60,13 @@ private ErrorHandlingUtils() {
 	 * @param logger the logger.
 	 * @param logLevel the log level.
 	 * @param retryListeners 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, List<RetryListener> retryListeners) {
+			KafkaException.Level logLevel, List<RetryListener> retryListeners, BinaryExceptionClassifier classifier) {
 
 		BackOffExecution execution = backOff.start();
 		long nextBackOff = execution.nextBackOff();
@@ -79,7 +82,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);
@@ -145,4 +149,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 cause) {
+			theEx = cause;
+		}
+		if (theEx instanceof ListenerExecutionFailedException && theEx.getCause() instanceof Exception cause) {
+			theEx = cause;
+		}
+		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;
@@ -149,6 +150,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;
@@ -101,6 +102,31 @@ public void setLogLevel(Level logLevel) {
 		}
 	}
 
+	@Override
+	protected void notRetryable(Stream<Class<? extends Exception>> notRetryable) {
+		if (this.fallbackBatchHandler instanceof ExceptionClassifier handler) {
+			notRetryable.forEach(ex -> handler.addNotRetryableExceptions(ex));
+		}
+	}
+
+	@Override
+	public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifications, boolean defaultValue) {
+		super.setClassifications(classifications, defaultValue);
+		if (this.fallbackBatchHandler instanceof ExceptionClassifier handler) {
+			handler.setClassifications(classifications, defaultValue);
+		}
+	}
+
+	@Override
+	@Nullable
+	public Boolean removeClassification(Class<? extends Exception> exceptionType) {
+		Boolean removed = super.removeClassification(exceptionType);
+		if (this.fallbackBatchHandler instanceof ExceptionClassifier handler) {
+			handler.removeClassification(exceptionType);
+		}
+		return removed;
+	}
+
 	/**
 	 * Return the fallback batch error handler.
 	 * @return the handler.

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

@@ -45,13 +45,7 @@ public abstract class FailedRecordProcessor extends ExceptionClassifier implemen
 
 	private final BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> noRetriesForClassified =
 			(rec, ex) -> {
-				Exception theEx = ex;
-				if (theEx instanceof TimestampedException && theEx.getCause() instanceof Exception cause) {
-					theEx = cause;
-				}
-				if (theEx instanceof ListenerExecutionFailedException && theEx.getCause() instanceof Exception cause) {
-					theEx = cause;
-				}
+				Exception theEx = ErrorHandlingUtils.unwrapIfNeeded(ex);
 				if (!getClassifier().classify(theEx)) {
 					return NO_RETRIES_OR_DELAY_BACKOFF;
 				}

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

@@ -48,7 +48,7 @@
  * @since 2.3.7
  *
  */
-class FallbackBatchErrorHandler extends KafkaExceptionLogLevelAware implements CommonErrorHandler {
+class FallbackBatchErrorHandler extends ExceptionClassifier implements CommonErrorHandler {
 
 	private final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
 
@@ -124,7 +124,7 @@ public void handleBatch(Exception thrownException, @Nullable ConsumerRecords<?,
 		this.retrying.set(true);
 		try {
 			ErrorHandlingUtils.retryBatch(thrownException, records, consumer, container, invokeListener, this.backOff,
-					this.seeker, this.recoverer, this.logger, getLogLevel(), this.retryListeners);
+					this.seeker, this.recoverer, this.logger, getLogLevel(), this.retryListeners, 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 {