From 0fdb5f49b5c33bceb27899ef123027a9a238ab00 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@vmware.com>
Date: Mon, 24 Oct 2022 15:40:28 -0400
Subject: [PATCH] GH-2427: Allow RuntimeException to be Classified

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

Previously, classifying `RuntimeException` either for no retries or for
blocking retries would cause undesirable effects - its classification
would be found first because the classifier first traverses up the
class hierarchy to find a match before traversing down the cause links.

When classifying exception for retry, unwrap the 'LEFE' cause from a
`TimestampedException` and/or `ListenerExecutionFailedException` so
that the classification is made on that cause.

By default, the retry classifier has no classified exceptions when
used in "classify for retry" mode (instead of the default "classify
for no retry" mode). This means that, if `retryOn(RuntimeException.class)`
is used, then all `RuntimeException`s will be retried (including those
that are usually considered fatal).

With mixed blocking/non-blocking retries, change the behavior to include
the standard fatal exceptions in case the user configures all `RuntimeException`s
to use blocking retries.

Also tested with a Boot app to see that a conversion exception bypasses
all retries and goes straight to the DLT.
---
 .../kafka/listener/ExceptionClassifier.java   | 25 +++++++++--
 .../kafka/listener/FailedRecordProcessor.java |  9 +++-
 .../ListenerContainerFactoryConfigurer.java   | 14 +++++-
 .../RetryTopicConfigurationSupport.java       | 44 +++++++++++++++----
 .../DeadLetterPublishingRecovererTests.java   | 19 ++++++++
 ...stenerContainerFactoryConfigurerTests.java | 37 ++++++++++++++++
 .../RetryTopicConfigurationSupportTests.java  |  1 +
 7 files changed, 135 insertions(+), 14 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java
index 43f23448b7..6bf67859b0 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ExceptionClassifier.java
@@ -45,7 +45,7 @@ public abstract class ExceptionClassifier extends KafkaExceptionLogLevelAware {
 	 * Construct the instance.
 	 */
 	public ExceptionClassifier() {
-		this.classifier = configureDefaultClassifier();
+		this.classifier = configureDefaultClassifier(true);
 	}
 
 	/**
@@ -64,9 +64,27 @@ public static List<Class<? extends Throwable>> defaultFatalExceptionsList() {
 							ClassCastException.class);
 	}
 
-	private static ExtendedBinaryExceptionClassifier configureDefaultClassifier() {
+	private static ExtendedBinaryExceptionClassifier configureDefaultClassifier(boolean defaultClassification) {
 		return new ExtendedBinaryExceptionClassifier(defaultFatalExceptionsList().stream()
-				.collect(Collectors.toMap(ex -> ex, ex -> false)), true);
+				.collect(Collectors.toMap(ex -> ex, ex -> false)), defaultClassification);
+	}
+
+	/**
+	 * By default, unmatched types classify as true. Call this method to make the default
+	 * false, and optionally retain types implicitly classified as false. This should be
+	 * called before calling any of the classification modification methods. This can be
+	 * useful if you want to classify a super class of one or more of the standard fatal
+	 * exceptions as retryable.
+	 * @param retainStandardFatal true to retain.
+	 * @since 3.0
+	 */
+	public void defaultFalse(boolean retainStandardFatal) {
+		if (retainStandardFatal) {
+			this.classifier = configureDefaultClassifier(false);
+		}
+		else {
+			defaultFalse();
+		}
 	}
 
 	/**
@@ -94,6 +112,7 @@ protected BinaryExceptionClassifier getClassifier() {
 	 * <ul>
 	 * <li>{@link DeserializationException}</li>
 	 * <li>{@link MessageConversionException}</li>
+	 * <li>{@link ConversionException}</li>
 	 * <li>{@link MethodArgumentResolutionException}</li>
 	 * <li>{@link NoSuchMethodException}</li>
 	 * <li>{@link ClassCastException}</li>
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
index 96b58607cf..76efc1fbe9 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
@@ -45,7 +45,14 @@ public abstract class FailedRecordProcessor extends ExceptionClassifier implemen
 
 	private final BiFunction<ConsumerRecord<?, ?>, Exception, BackOff> noRetriesForClassified =
 			(rec, ex) -> {
-				if (!getClassifier().classify(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;
+				}
+				if (!getClassifier().classify(theEx)) {
 					return NO_RETRIES_OR_DELAY_BACKOFF;
 				}
 				return this.userBackOffFunction.apply(rec, ex);
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java
index 88ae8e4295..1b187e45f9 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java
@@ -62,6 +62,8 @@ public class ListenerContainerFactoryConfigurer {
 
 	private Class<? extends Exception>[] blockingExceptionTypes = null;
 
+	private boolean retainStandardFatal;
+
 	private Consumer<ConcurrentMessageListenerContainer<?, ?>> containerCustomizer = container -> {
 	};
 
@@ -141,6 +143,16 @@ public final void setBlockingRetryableExceptions(Class<? extends Exception>... e
 		this.blockingExceptionTypes = Arrays.copyOf(exceptionTypes, exceptionTypes.length);
 	}
 
+	/**
+	 * Set to true to retain standard fatal exceptions as not retryable when configuring
+	 * blocking retries.
+	 * @param retainStandardFatal true to retain standard fatal exceptions.
+	 * @since 3.0
+	 */
+	public void setRetainStandardFatal(boolean retainStandardFatal) {
+		this.retainStandardFatal = retainStandardFatal;
+	}
+
 	public void setContainerCustomizer(Consumer<ConcurrentMessageListenerContainer<?, ?>> containerCustomizer) {
 		Assert.notNull(containerCustomizer, "'containerCustomizer' cannot be null");
 		this.containerCustomizer = containerCustomizer;
@@ -153,7 +165,7 @@ public void setErrorHandlerCustomizer(Consumer<DefaultErrorHandler> errorHandler
 	protected CommonErrorHandler createErrorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer,
 												Configuration configuration) {
 		DefaultErrorHandler errorHandler = createDefaultErrorHandlerInstance(deadLetterPublishingRecoverer);
-		errorHandler.defaultFalse();
+		errorHandler.defaultFalse(this.retainStandardFatal);
 		errorHandler.setCommitRecovered(true);
 		errorHandler.setLogLevel(KafkaException.Level.DEBUG);
 		if (this.blockingExceptionTypes != null) {
diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupport.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupport.java
index b2ae0e62e1..4079bf09f7 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupport.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupport.java
@@ -141,7 +141,7 @@ private void processDeadLetterPublishingContainerFactory(
 		CustomizersConfigurer customizersConfigurer = new CustomizersConfigurer();
 		configureCustomizers(customizersConfigurer);
 		JavaUtils.INSTANCE
-				.acceptIfNotNull(customizersConfigurer.deadLetterPublishingRecovererCustomizer,
+				.acceptIfNotNull(customizersConfigurer.getDeadLetterPublishingRecovererCustomizer(),
 						deadLetterPublishingRecovererFactory::setDeadLetterPublishingRecovererCustomizer);
 		Consumer<DeadLetterPublishingRecovererFactory> dlprfConsumer = configureDeadLetterPublishingContainerFactory();
 		Assert.notNull(dlprfConsumer, "configureDeadLetterPublishingContainerFactory must not return null");
@@ -159,24 +159,28 @@ protected Consumer<DeadLetterPublishingRecovererFactory> configureDeadLetterPubl
 
 	/**
 	 * Internal method for processing the {@link ListenerContainerFactoryConfigurer}.
-	 * Consider overriding {@link #configureListenerContainerFactoryConfigurer()}
-	 * if further customization is required.
-	 * @param listenerContainerFactoryConfigurer the {@link ListenerContainerFactoryConfigurer} instance.
+	 * Consider overriding {@link #configureListenerContainerFactoryConfigurer()} if
+	 * further customization is required.
+	 * @param listenerContainerFactoryConfigurer the
+	 * {@link ListenerContainerFactoryConfigurer} instance.
 	 */
-	private void processListenerContainerFactoryConfigurer(ListenerContainerFactoryConfigurer listenerContainerFactoryConfigurer) {
+	private void processListenerContainerFactoryConfigurer(
+			ListenerContainerFactoryConfigurer listenerContainerFactoryConfigurer) {
+
 		CustomizersConfigurer customizersConfigurer = new CustomizersConfigurer();
 		configureCustomizers(customizersConfigurer);
 		BlockingRetriesConfigurer blockingRetriesConfigurer = new BlockingRetriesConfigurer();
 		configureBlockingRetries(blockingRetriesConfigurer);
 		JavaUtils.INSTANCE
-				.acceptIfNotNull(blockingRetriesConfigurer.backOff,
+				.acceptIfNotNull(blockingRetriesConfigurer.getBackOff(),
 						listenerContainerFactoryConfigurer::setBlockingRetriesBackOff)
-				.acceptIfNotNull(blockingRetriesConfigurer.retryableExceptions,
+				.acceptIfNotNull(blockingRetriesConfigurer.getRetryableExceptions(),
 						listenerContainerFactoryConfigurer::setBlockingRetryableExceptions)
-				.acceptIfNotNull(customizersConfigurer.errorHandlerCustomizer,
+				.acceptIfNotNull(customizersConfigurer.getErrorHandlerCustomizer(),
 						listenerContainerFactoryConfigurer::setErrorHandlerCustomizer)
-				.acceptIfNotNull(customizersConfigurer.listenerContainerCustomizer,
+				.acceptIfNotNull(customizersConfigurer.getListenerContainerCustomizer(),
 						listenerContainerFactoryConfigurer::setContainerCustomizer);
+		listenerContainerFactoryConfigurer.setRetainStandardFatal(true);
 		Consumer<ListenerContainerFactoryConfigurer> lcfcConfigurer = configureListenerContainerFactoryConfigurer();
 		Assert.notNull(lcfcConfigurer, "configureListenerContainerFactoryConfigurer must not return null.");
 		lcfcConfigurer.accept(listenerContainerFactoryConfigurer);
@@ -341,6 +345,15 @@ public BlockingRetriesConfigurer backOff(BackOff backoff) {
 			this.backOff = backoff;
 			return this;
 		}
+
+		BackOff getBackOff() {
+			return this.backOff;
+		}
+
+		Class<? extends Exception>[] getRetryableExceptions() {
+			return this.retryableExceptions;
+		}
+
 	}
 
 	/**
@@ -387,6 +400,19 @@ public CustomizersConfigurer customizeDeadLetterPublishingRecoverer(Consumer<Dea
 			this.deadLetterPublishingRecovererCustomizer = dlprCustomizer;
 			return this;
 		}
+
+		Consumer<DefaultErrorHandler> getErrorHandlerCustomizer() {
+			return this.errorHandlerCustomizer;
+		}
+
+		Consumer<ConcurrentMessageListenerContainer<?, ?>> getListenerContainerCustomizer() {
+			return this.listenerContainerCustomizer;
+		}
+
+		Consumer<DeadLetterPublishingRecoverer> getDeadLetterPublishingRecovererCustomizer() {
+			return this.deadLetterPublishingRecovererCustomizer;
+		}
+
 	}
 
 }
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java
index aba4126f67..5cc4a4f513 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/DeadLetterPublishingRecovererTests.java
@@ -70,6 +70,7 @@
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer.HeaderNames;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.SendResult;
+import org.springframework.kafka.support.converter.ConversionException;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.kafka.support.serializer.SerializationUtils;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -898,4 +899,22 @@ void nonCompliantProducerFactory() throws Exception {
 		assertThat(timeoutCaptor.getValue()).isEqualTo(Duration.ofSeconds(125).toMillis());
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void blockingRetryRuntimeException() {
+		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
+		CompletableFuture future = mock(CompletableFuture.class);
+		given(template.send(any(ProducerRecord.class))).willReturn(future);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		recoverer.defaultFalse(true);
+		recoverer.addRetryableExceptions(RuntimeException.class);
+		recoverer.accept(record, new ListenerExecutionFailedException("test", "group",
+				new TimestampedException(
+						new ListenerExecutionFailedException("test", new ConversionException("test", null)))));
+		ArgumentCaptor<ProducerRecord> producerRecordCaptor = ArgumentCaptor.forClass(ProducerRecord.class);
+		verify(template).send(producerRecordCaptor.capture());
+		ProducerRecord outRecord = producerRecordCaptor.getValue();
+	}
+
 }
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java
index dd7cd0a9a3..6404f9b27c 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurerTests.java
@@ -68,6 +68,7 @@
 
 /**
  * @author Tomaz Fernandes
+ * @author Gary Russell
  * @since 2.7
  */
 @ExtendWith(MockitoExtension.class)
@@ -295,6 +296,42 @@ void shouldUseGivenBackOffAndExceptions() {
 
 	}
 
+	@Test
+	void shouldUseGivenBackOffAndExceptionsKeepStandard() {
+
+		// given
+		given(container.getContainerProperties()).willReturn(containerProperties);
+		given(deadLetterPublishingRecovererFactory.create()).willReturn(recoverer);
+		given(containerProperties.getMessageListener()).willReturn(listener);
+		given(configuration.forContainerFactoryConfigurer()).willReturn(lcfcConfiguration);
+		willReturn(container).given(containerFactory).createListenerContainer(endpoint);
+		BackOff backOffMock = mock(BackOff.class);
+		BackOffExecution backOffExecutionMock = mock(BackOffExecution.class);
+		given(backOffMock.start()).willReturn(backOffExecutionMock);
+
+		ListenerContainerFactoryConfigurer configurer =
+				new ListenerContainerFactoryConfigurer(kafkaConsumerBackoffManager,
+						deadLetterPublishingRecovererFactory, clock);
+		configurer.setBlockingRetriesBackOff(backOffMock);
+		configurer.setBlockingRetryableExceptions(IllegalArgumentException.class, IllegalStateException.class);
+		configurer.setRetainStandardFatal(true);
+
+		// when
+		KafkaListenerContainerFactory<?> decoratedFactory =
+				configurer.decorateFactory(this.containerFactory, configuration.forContainerFactoryConfigurer());
+		decoratedFactory.createListenerContainer(endpoint);
+
+		// then
+		then(backOffMock).should().start();
+		then(container).should().setCommonErrorHandler(errorHandlerCaptor.capture());
+		CommonErrorHandler errorHandler = errorHandlerCaptor.getValue();
+		assertThat(DefaultErrorHandler.class.isAssignableFrom(errorHandler.getClass())).isTrue();
+		DefaultErrorHandler defaultErrorHandler = (DefaultErrorHandler) errorHandler;
+		assertThat(defaultErrorHandler.removeClassification(IllegalArgumentException.class)).isTrue();
+		assertThat(defaultErrorHandler.removeClassification(IllegalStateException.class)).isTrue();
+		assertThat(defaultErrorHandler.removeClassification(ConversionException.class)).isFalse();
+
+	}
 
 	@Test
 	void shouldThrowIfBackOffOrRetryablesAlreadySet() {
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupportTests.java b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupportTests.java
index 0f072ed997..703d74691a 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupportTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationSupportTests.java
@@ -143,6 +143,7 @@ protected void configureBlockingRetries(BlockingRetriesConfigurer blockingRetrie
 		then(lcfc).should().setErrorHandlerCustomizer(errorHandlerCustomizer);
 		assertThatThrownBy(lcfc::setBlockingRetryableExceptions).isInstanceOf(IllegalStateException.class);
 		then(lcfc).should().setBlockingRetriesBackOff(backoff);
+		then(lcfc).should().setRetainStandardFatal(true);
 		then(dlprfCustomizer).should().accept(dlprf);
 		then(rtconfigurer).should().accept(topicConfigurer);
 		then(lcfcConsumer).should().accept(lcfc);