From ce9656b27fd64cbfb71dcedacb2d68146b3a74aa Mon Sep 17 00:00:00 2001
From: Tomaz Fernandes <tomaz.fernandes.se@gmail.com>
Date: Thu, 24 Feb 2022 12:45:06 -0300
Subject: [PATCH] Improve ExceptionClassifier JavaDoc

Also add assertions to the LCFC new methods to warn the user if they already set the blocking configurations.
---
 .../kafka/listener/ExceptionClassifier.java   |  3 ++-
 .../ListenerContainerFactoryConfigurer.java   |  9 ++++++++
 ...stenerContainerFactoryConfigurerTests.java | 23 ++++++++++++++++++-
 3 files changed, 33 insertions(+), 2 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 281bc7fc6a..723b4949bf 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
@@ -181,7 +181,8 @@ public boolean removeNotRetryableException(Class<? extends Exception> exceptionT
 	 * </ul>
 	 * All others will be retried, unless {@link #defaultFalse()} has been called.
 	 * @param exceptionType the exception type.
-	 * @return true if the removal was successful.
+	 * @return the classification of the exception if removal was successful;
+	 * null otherwise.
 	 * @since 2.8.4
 	 * @see #addNotRetryableExceptions(Class...)
 	 * @see #setClassifications(Map, boolean)
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 ba3816d52a..b1793b2492 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
@@ -17,6 +17,7 @@
 package org.springframework.kafka.retrytopic;
 
 import java.time.Clock;
+import java.util.Arrays;
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
@@ -173,6 +174,10 @@ public KafkaListenerContainerFactory<?> decorateFactoryWithoutSettingContainerPr
 	 */
 	public void setBlockingRetriesBackOff(BackOff blockingBackOff) {
 		Assert.notNull(blockingBackOff, "The provided BackOff cannot be null");
+		Assert.state(this.providedBlockingBackOff == null, () ->
+				"Blocking retries back off has already been set. Current: "
+						+ this.providedBlockingBackOff
+						+ " You provided: " + blockingBackOff);
 		this.providedBlockingBackOff = blockingBackOff;
 	}
 
@@ -187,6 +192,10 @@ public void setBlockingRetriesBackOff(BackOff blockingBackOff) {
 	public final void setBlockingRetryableExceptions(Class<? extends Exception>... exceptionTypes) {
 		Assert.notNull(exceptionTypes, "The exception types cannot be null");
 		Assert.noNullElements(exceptionTypes, "The exception types cannot have null elements");
+		Assert.state(this.blockingExceptionTypes == null,
+				() -> "Blocking retryable exceptions have already been set."
+						+  "Current ones: " + Arrays.toString(this.blockingExceptionTypes)
+						+ " You provided: " + Arrays.toString(exceptionTypes));
 		this.blockingExceptionTypes = exceptionTypes;
 	}
 
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 750844c339..48f24c19a2 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
@@ -17,6 +17,7 @@
 package org.springframework.kafka.retrytopic;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.eq;
@@ -61,15 +62,17 @@
 import org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.converter.ConversionException;
+import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.BackOffExecution;
+import org.springframework.util.backoff.FixedBackOff;
 
 /**
  * @author Tomaz Fernandes
  * @since 2.7
  */
 @ExtendWith(MockitoExtension.class)
-@SuppressWarnings({"unchecked", "rawtypes"})
+@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
 class ListenerContainerFactoryConfigurerTests {
 
 	@Mock
@@ -447,6 +450,24 @@ void shouldUseGivenBackOffAndExceptions() {
 
 	}
 
+
+	@Test
+	void shouldThrowIfBackOffOrRetryablesAlreadySet() {
+		// given
+		BackOff backOff = new FixedBackOff();
+		ListenerContainerFactoryConfigurer configurer =
+				new ListenerContainerFactoryConfigurer(kafkaConsumerBackoffManager,
+						deadLetterPublishingRecovererFactory, clock);
+		configurer.setBlockingRetriesBackOff(backOff);
+		configurer.setBlockingRetryableExceptions(IllegalArgumentException.class, IllegalStateException.class);
+
+		// when / then
+		assertThatThrownBy(() -> configurer.setBlockingRetriesBackOff(backOff)).isInstanceOf(IllegalStateException.class);
+		assertThatThrownBy(() -> configurer.setBlockingRetryableExceptions(ConversionException.class, DeserializationException.class))
+				.isInstanceOf(IllegalStateException.class);
+	}
+
+
 	@Test
 	void shouldCacheFactoryInstances() {