GH-1918: Don't Retry DLT Fatal Exceptions

Resolves #1918

Do not publish records with fatal exceptions to the same topic, it
causes an irrecoverable endless loop.

Log at ERROR when such a situation is detected.

Extract a new abstract super class for exception classification.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

@@ -532,11 +532,11 @@ public class MyCustomDltProcessor {
 
 NOTE: If no DLT handler is provided, the default RetryTopicConfigurer.LoggingDltListenerHandlerMethod is used.
 
-===== Dlt Failure Behavior
+===== DLT Failure Behavior
 
-Should the Dlt processing fail, there are two possible behaviors available: `ALWAYS_RETRY_ON_ERROR` and `FAIL_ON_ERROR`.
+Should the DLT processing fail, there are two possible behaviors available: `ALWAYS_RETRY_ON_ERROR` and `FAIL_ON_ERROR`.
 
-In the former the message is forwarded back to the dlt topic so it doesn't block other dlt messages' processing.
+In the former the record is forwarded back to the DLT topic so it doesn't block other DLT records' processing.
 In the latter the consumer ends the execution without forwarding the message.
 
 ====
@@ -566,7 +566,21 @@ public RetryTopicConfiguration myRetryTopic(KafkaTemplate<Integer, MyPojo> templ
 
 NOTE: The default behavior is to `ALWAYS_RETRY_ON_ERROR`.
 
-===== Configuring No Dlt
+IMPORTANT: Starting with version 2.8, `ALWAYS_RETRY_ON_ERROR` will NOT route a record back to the DLT if the record causes a fatal exception to be thrown,
+such as a `DeserializationException` because, generally, such exceptions will always be thrown.
+
+Exceptions that are considered fatal are
+
+* `DeserializationException`
+* `MessageConversionException`
+* `ConversionException`
+* `MethodArgumentResolutionException`
+* `NoSuchMethodException`
+* `ClassCastException`
+
+You can add exceptions to and remove exceptions from this list using methods on the `DeadLetterPublishingRecovererFactory` bean.
+
+===== Configuring No DLT
 
 The framework also provides the possibility of not configuring a DLT for the topic.
 In this case after retrials are exhausted the processing simply ends.

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

@@ -64,7 +64,7 @@
  * @since 2.2
  *
  */
-public class DeadLetterPublishingRecoverer implements ConsumerAwareRecordRecoverer {
+public class DeadLetterPublishingRecoverer extends ExceptionClassifier implements ConsumerAwareRecordRecoverer {
 
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
 
@@ -305,6 +305,14 @@ public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consume
 		TopicPartition tp = this.destinationResolver.apply(record, exception);
 		if (tp == null) {
 			maybeThrow(record, exception);
+			this.logger.debug(() -> "Recovery of " + ListenerUtils.recordToString(record, true)
+					+ " skipped because destination resolver returned null");
+			return;
+		}
+		if (tp.topic().equals(record.topic()) && !getClassifier().classify(exception)) {
+			this.logger.error("Recovery of " + ListenerUtils.recordToString(record, true)
+					+ " skipped because not retryable exception " + exception.toString()
+					+ " and the destination resolver routed back to the same topic");
 			return;
 		}
 		if (consumer != null && this.verifyPartition) {
@@ -320,7 +328,7 @@ public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consume
 				kDeserEx == null ? null : kDeserEx.getData(), vDeserEx == null ? null : vDeserEx.getData());
 		KafkaOperations<Object, Object> kafkaTemplate =
 				(KafkaOperations<Object, Object>) this.templateResolver.apply(outRecord);
-		sendOrThrow(outRecord, kafkaTemplate);
+		sendOrThrow(outRecord, kafkaTemplate, record);
 	}
 
 	private void addAndEnhanceHeaders(ConsumerRecord<?, ?> record, Exception exception,
@@ -345,10 +353,10 @@ private void addAndEnhanceHeaders(ConsumerRecord<?, ?> record, Exception excepti
 	}
 
 	private void sendOrThrow(ProducerRecord<Object, Object> outRecord,
-			@Nullable KafkaOperations<Object, Object> kafkaTemplate) {
+			@Nullable KafkaOperations<Object, Object> kafkaTemplate, ConsumerRecord<?, ?> inRecord) {
 
 		if (kafkaTemplate != null) {
-			send(outRecord, kafkaTemplate);
+			send(outRecord, kafkaTemplate, inRecord);
 		}
 		else {
 			throw new IllegalArgumentException("No kafka template returned for record " + outRecord);
@@ -369,17 +377,20 @@ private void maybeThrow(ConsumerRecord<?, ?> record, Exception exception) {
 	 * Send the record.
 	 * @param outRecord the record.
 	 * @param kafkaTemplate the template.
+	 * @param inRecord the consumer record.
 	 * @since 2.7
 	 */
-	protected void send(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate) {
+	protected void send(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate,
+			ConsumerRecord<?, ?> inRecord) {
+
 		if (this.transactional && !kafkaTemplate.inTransaction() && !kafkaTemplate.isAllowNonTransactional()) {
 			kafkaTemplate.executeInTransaction(t -> {
-				publish(outRecord, t);
+				publish(outRecord, t, inRecord);
 				return null;
 			});
 		}
 		else {
-			publish(outRecord, kafkaTemplate);
+			publish(outRecord, kafkaTemplate, inRecord);
 		}
 	}
 
@@ -409,7 +420,8 @@ private TopicPartition checkPartition(TopicPartition tp, Consumer<?, ?> consumer
 
 	@SuppressWarnings("unchecked")
 	private KafkaOperations<Object, Object> findTemplateForValue(@Nullable Object value,
-																Map<Class<?>, KafkaOperations<?, ?>> templates) {
+			Map<Class<?>, KafkaOperations<?, ?>> templates) {
+
 		if (value == null) {
 			KafkaOperations<?, ?> operations = templates.get(Void.class);
 			if (operations == null) {
@@ -461,46 +473,55 @@ protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?,
 	 * Override this if you want more than just logging of the send result.
 	 * @param outRecord the record to send.
 	 * @param kafkaTemplate the template.
+	 * @param inRecord the consumer record.
 	 * @since 2.2.5
 	 */
-	protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate) {
+	protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate,
+			ConsumerRecord<?, ?> inRecord) {
+
 		ListenableFuture<SendResult<Object, Object>> sendResult = null;
 		try {
 			sendResult = kafkaTemplate.send(outRecord);
 			sendResult.addCallback(result -> {
-				this.logger.debug(() -> "Successful dead-letter publication: " + result);
+				this.logger.debug(() -> "Successful dead-letter publication: "
+						+ ListenerUtils.recordToString(inRecord, true) + " to " + result.getRecordMetadata());
 			}, ex -> {
-				this.logger.error(ex, () -> "Dead-letter publication failed for: " + outRecord);
+				this.logger.error(ex, () -> pubFailMessage(outRecord, inRecord));
 			});
 		}
 		catch (Exception e) {
-			this.logger.error(e, () -> "Dead-letter publication failed for: " + outRecord);
+			this.logger.error(e, () -> pubFailMessage(outRecord, inRecord));
 		}
 		if (this.failIfSendResultIsError) {
-			verifySendResult(kafkaTemplate, outRecord, sendResult);
+			verifySendResult(kafkaTemplate, outRecord, sendResult, inRecord);
 		}
 	}
 
 	private void verifySendResult(KafkaOperations<Object, Object> kafkaTemplate,
 			ProducerRecord<Object, Object> outRecord,
-			@Nullable ListenableFuture<SendResult<Object, Object>> sendResult) {
+			@Nullable ListenableFuture<SendResult<Object, Object>> sendResult, ConsumerRecord<?, ?> inRecord) {
 
 		Duration sendTimeout = determineSendTimeout(kafkaTemplate);
 		if (sendResult == null) {
-			throw new KafkaException("Dead-letter publication failed for: " + outRecord);
+			throw new KafkaException(pubFailMessage(outRecord, inRecord));
 		}
 		try {
 			sendResult.get(sendTimeout.toMillis(), TimeUnit.MILLISECONDS);
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
-			throw new KafkaException("Publication failed for: " + outRecord, e);
+			throw new KafkaException(pubFailMessage(outRecord, inRecord));
 		}
 		catch (ExecutionException | TimeoutException e) {
-			throw new KafkaException("Publication failed for: " + outRecord, e);
+			throw new KafkaException(pubFailMessage(outRecord, inRecord), e);
 		}
 	}
 
+	private String pubFailMessage(ProducerRecord<Object, Object> outRecord, ConsumerRecord<?, ?> inRecord) {
+		return "Dead-letter publication to "
+				+ outRecord.topic() + "failed for: " + ListenerUtils.recordToString(inRecord, true);
+	}
+
 	private Duration determineSendTimeout(KafkaOperations<?, ?> template) {
 		ProducerFactory<? extends Object, ? extends Object> producerFactory = template.getProducerFactory();
 		if (producerFactory != null) { // NOSONAR - will only occur in mock tests

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

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2021 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 java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.classify.BinaryExceptionClassifier;
+import org.springframework.kafka.support.converter.ConversionException;
+import org.springframework.kafka.support.serializer.DeserializationException;
+import org.springframework.messaging.converter.MessageConversionException;
+import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException;
+import org.springframework.util.Assert;
+
+/**
+ * Supports exception classification.
+ *
+ * @author Gary Russell
+ * @since 2.8
+ *
+ */
+public abstract class ExceptionClassifier extends KafkaExceptionLogLevelAware {
+
+	private ExtendedBinaryExceptionClassifier classifier;
+
+	/**
+	 * Construct the instance.
+	 */
+	public ExceptionClassifier() {
+		this.classifier = configureDefaultClassifier();
+	}
+
+	private static ExtendedBinaryExceptionClassifier configureDefaultClassifier() {
+		Map<Class<? extends Throwable>, Boolean> classified = new HashMap<>();
+		classified.put(DeserializationException.class, false);
+		classified.put(MessageConversionException.class, false);
+		classified.put(ConversionException.class, false);
+		classified.put(MethodArgumentResolutionException.class, false);
+		classified.put(NoSuchMethodException.class, false);
+		classified.put(ClassCastException.class, false);
+		return new ExtendedBinaryExceptionClassifier(classified, true);
+	}
+
+	/**
+	 * Return the exception classifier.
+	 * @return the classifier.
+	 */
+	protected BinaryExceptionClassifier getClassifier() {
+		return this.classifier;
+	}
+
+	/**
+	 * Set an exception classifications to determine whether the exception should cause a retry
+	 * (until exhaustion) or not. If not, we go straight to the recoverer. By default,
+	 * the following exceptions will not be retried:
+	 * <ul>
+	 * <li>{@link DeserializationException}</li>
+	 * <li>{@link MessageConversionException}</li>
+	 * <li>{@link MethodArgumentResolutionException}</li>
+	 * <li>{@link NoSuchMethodException}</li>
+	 * <li>{@link ClassCastException}</li>
+	 * </ul>
+	 * All others will be retried.
+	 * When calling this method, the defaults will not be applied.
+	 * @param classifications the classifications.
+	 * @param defaultValue whether or not to retry non-matching exceptions.
+	 * @see BinaryExceptionClassifier#BinaryExceptionClassifier(Map, boolean)
+	 * @see #addNotRetryableExceptions(Class...)
+	 */
+	public void setClassifications(Map<Class<? extends Throwable>, Boolean> classifications, boolean defaultValue) {
+		Assert.notNull(classifications, "'classifications' + cannot be null");
+		this.classifier = new ExtendedBinaryExceptionClassifier(classifications, defaultValue);
+	}
+
+	/**
+	 * Add exception types to the default list. By default, the following exceptions will
+	 * not be retried:
+	 * <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>
+	 * </ul>
+	 * All others will be retried.
+	 * @param exceptionTypes the exception types.
+	 * @see #removeNotRetryableException(Class)
+	 * @see #setClassifications(Map, boolean)
+	 */
+	@SafeVarargs
+	@SuppressWarnings("varargs")
+	public final void addNotRetryableExceptions(Class<? extends Exception>... exceptionTypes) {
+		Assert.notNull(exceptionTypes, "'exceptionTypes' cannot be null");
+		Assert.noNullElements(exceptionTypes, "'exceptionTypes' cannot contain nulls");
+		for (Class<? extends Exception> exceptionType : exceptionTypes) {
+			Assert.isTrue(Exception.class.isAssignableFrom(exceptionType),
+					() -> "exceptionType " + exceptionType + " must be an Exception");
+			this.classifier.getClassified().put(exceptionType, false);
+		}
+	}
+
+	/**
+	 * Remove an exception type from the configured list. By default, the following
+	 * exceptions will not be retried:
+	 * <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>
+	 * </ul>
+	 * All others will be retried.
+	 * @param exceptionType the exception type.
+	 * @return true if the removal was successful.
+	 * @see #addNotRetryableExceptions(Class...)
+	 * @see #setClassifications(Map, boolean)
+	 */
+	public boolean removeNotRetryableException(Class<? extends Exception> exceptionType) {
+		return this.classifier.getClassified().remove(exceptionType);
+	}
+
+	/**
+	 * Extended to provide visibility to the current classified exceptions.
+	 *
+	 * @author Gary Russell
+	 *
+	 */
+	@SuppressWarnings("serial")
+	private static final class ExtendedBinaryExceptionClassifier extends BinaryExceptionClassifier {
+
+		ExtendedBinaryExceptionClassifier(Map<Class<? extends Throwable>, Boolean> typeMap, boolean defaultValue) {
+			super(typeMap, defaultValue);
+			setTraverseCauses(true);
+		}
+
+		@Override
+		protected Map<Class<? extends Throwable>, Boolean> getClassified() { // NOSONAR worthless override
+			return super.getClassified();
+		}
+
+	}
+
+}