GH-2118: Make DLPR skip same topic fatal ex optional

Fixes GH-2118

The logic introduced to stop endless loops in DLR processing for fatal exceptions interferes with RT's single-topic strategy.
Make it optional and set the default to false in DLPRF for RT.
The loop is now addressed in DefaultDestinationTopicResolver as in GH-2113.

Author: tomazfernandes

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

@@ -103,6 +103,8 @@ public class DeadLetterPublishingRecoverer extends ExceptionClassifier implement
 
 	private boolean stripPreviousExceptionHeaders = true;
 
+	private boolean skipSameTopicFatalExceptions = true;
+
 	/**
 	 * Create an instance with the provided template and a default destination resolving
 	 * function that returns a TopicPartition based on the original topic (appended with ".DLT")
@@ -327,6 +329,16 @@ public void setStripPreviousExceptionHeaders(boolean stripPreviousExceptionHeade
 		this.stripPreviousExceptionHeaders = stripPreviousExceptionHeaders;
 	}
 
+	/**
+	 * Set to false if you want to forward the record to the same topic even though
+	 * the exception is fatal by this class' classification, e.g. to handle this scenario
+	 * in a different layer.
+	 * @param skipSameTopicFatalExceptions false to forward in this scenario.
+	 */
+	public void setSkipSameTopicFatalExceptions(boolean skipSameTopicFatalExceptions) {
+		this.skipSameTopicFatalExceptions = skipSameTopicFatalExceptions;
+	}
+
 	@SuppressWarnings("unchecked")
 	@Override
 	public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consumer, Exception exception) {
@@ -337,7 +349,9 @@ public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consume
 					+ " skipped because destination resolver returned null");
 			return;
 		}
-		if (tp.topic().equals(record.topic()) && !getClassifier().classify(exception)) {
+		if (this.skipSameTopicFatalExceptions
+				&& 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");

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -140,9 +140,10 @@ protected DeadLetterPublishingRecoverer.HeaderNames getHeaderNames() {
 		recoverer.setFailIfSendResultIsError(true);
 		recoverer.setAppendOriginalHeaders(false);
 		recoverer.setThrowIfNoDestinationReturned(false);
+		recoverer.setSkipSameTopicFatalExceptions(false);
 		this.recovererCustomizer.accept(recoverer);
-		this.fatalExceptions.forEach(ex -> recoverer.addNotRetryableExceptions(ex));
-		this.nonFatalExceptions.forEach(ex -> recoverer.removeNotRetryableException(ex));
+		this.fatalExceptions.forEach(recoverer::addNotRetryableExceptions);
+		this.nonFatalExceptions.forEach(recoverer::removeNotRetryableException);
 		return recoverer;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -565,4 +565,25 @@ void noCircularRoutingIfFatal() {
 		verify(template).send(any(ProducerRecord.class));
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	void doNotSkipCircularFatalIfSet() {
+		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
+		ListenableFuture<Object> future = mock(ListenableFuture.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,
+				(cr, e) -> new TopicPartition("foo", 0));
+		recoverer.setSkipSameTopicFatalExceptions(false);
+		recoverer.accept(record, new ClassCastException());
+		verify(template).send(any(ProducerRecord.class));
+		recoverer.addNotRetryableExceptions(IllegalStateException.class);
+		recoverer.accept(record, new IllegalStateException());
+		verify(template, times(2)).send(any(ProducerRecord.class));
+		recoverer.removeNotRetryableException(IllegalStateException.class);
+		recoverer.setFailIfSendResultIsError(false);
+		recoverer.accept(record, new IllegalStateException());
+		verify(template, times(3)).send(any(ProducerRecord.class));
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DeadLetterPublishingRecovererFactoryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -306,14 +306,16 @@ void shouldCallDLPRCustomizer() {
 	}
 
 	@Test
-	void shouldNotSendMessageIfCircularFatal() {
+	void shouldSendMessageEvenIfCircularFatal() {
 		// setup
 		TimestampedException e = new TimestampedException(new IllegalStateException(), this.clock);
 		long failureTimestamp = e.getTimestamp();
 		given(destinationTopicResolver.resolveDestinationTopic(testTopic, 1, e, failureTimestamp))
 				.willReturn(destinationTopic);
 		given(destinationTopic.isNoOpsTopic()).willReturn(false);
 		given(destinationTopic.getDestinationName()).willReturn(testTopic);
+		given(this.destinationTopicResolver.getDestinationTopicByName(testTopic)).willReturn(destinationTopic);
+		willReturn(kafkaOperations).given(destinationTopic).getKafkaOperations();
 		this.consumerRecord.headers().add(RetryTopicHeaders.DEFAULT_HEADER_ORIGINAL_TIMESTAMP, originalTimestampBytes);
 
 		DeadLetterPublishingRecovererFactory factory = new DeadLetterPublishingRecovererFactory(
@@ -326,7 +328,7 @@ void shouldNotSendMessageIfCircularFatal() {
 		deadLetterPublishingRecoverer.accept(this.consumerRecord, e);
 
 		// then
-		then(kafkaOperations).should(times(0)).send(any(ProducerRecord.class));
+		then(kafkaOperations).should(times(1)).send(any(ProducerRecord.class));
 	}
 
 }