Improve ExceptionClassifier JavaDoc

Also add assertions to the LCFC new methods to warn the user if they already set the blocking configurations.

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)

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;
 	}
 

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() {