FailedRecordProcessor Polishing

- remove unnecessary assertions - it is no longer possible to set the classifier
- add varargs version of `addNotRetryableExceptions`

* Fix assertion in test.

Author: garyrussell

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

@@ -52,7 +52,7 @@ public abstract class FailedRecordProcessor extends KafkaExceptionLogLevelAware
 
 	private final FailedRecordTracker failureTracker;
 
-	private BinaryExceptionClassifier classifier;
+	private ExtendedBinaryExceptionClassifier classifier;
 
 	private boolean commitRecovered;
 
@@ -85,6 +85,7 @@ protected BinaryExceptionClassifier getClassifier() {
 	 * @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");
@@ -123,8 +124,8 @@ public int deliveryAttempt(TopicPartitionOffset topicPartitionOffset) {
 	}
 
 	/**
-	 * Add an exception type to the default list; if and only if an external classifier
-	 * has not been provided. By default, the following exceptions will not be retried:
+	 * Add an exception type to the default list. By default, the following exceptions
+	 * will not be retried:
 	 * <ul>
 	 * <li>{@link DeserializationException}</li>
 	 * <li>{@link MessageConversionException}</li>
@@ -134,19 +135,47 @@ public int deliveryAttempt(TopicPartitionOffset topicPartitionOffset) {
 	 * </ul>
 	 * All others will be retried.
 	 * @param exceptionType the exception type.
+	 * @deprecated in favor of {@link #addNotRetryableExceptions(Class...)}.
 	 * @see #removeNotRetryableException(Class)
 	 * @see #setClassifications(Map, boolean)
 	 */
+	@Deprecated
 	public void addNotRetryableException(Class<? extends Exception> exceptionType) {
-		Assert.isTrue(this.classifier instanceof ExtendedBinaryExceptionClassifier,
-				"Cannot add exception types to a supplied classifier");
-		((ExtendedBinaryExceptionClassifier) this.classifier).getClassified().put(exceptionType, false);
+		Assert.notNull(exceptionType, "'exceptionType' cannot be null");
+		this.classifier.getClassified().put(exceptionType, false);
 	}
 
 	/**
-	 * Remove an exception type from the configured list; if and only if an external
-	 * classifier has not been provided. By default, the following exceptions will not be
-	 * retried:
+	 * 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 MethodArgumentResolutionException}</li>
+	 * <li>{@link NoSuchMethodException}</li>
+	 * <li>{@link ClassCastException}</li>
+	 * </ul>
+	 * All others will be retried.
+	 * @param exceptionTypes the exception types.
+	 * @since 2.6
+	 * @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>
@@ -157,13 +186,11 @@ public void addNotRetryableException(Class<? extends Exception> exceptionType) {
 	 * All others will be retried.
 	 * @param exceptionType the exception type.
 	 * @return true if the removal was successful.
-	 * @see #addNotRetryableException(Class)
+	 * @see #addNotRetryableExceptions(Class...)
 	 * @see #setClassifications(Map, boolean)
 	 */
 	public boolean removeNotRetryableException(Class<? extends Exception> exceptionType) {
-		Assert.isTrue(this.classifier instanceof ExtendedBinaryExceptionClassifier,
-				"Cannot remove exception types from a supplied classifier");
-		return ((ExtendedBinaryExceptionClassifier) this.classifier).getClassified().remove(exceptionType);
+		return this.classifier.getClassified().remove(exceptionType);
 	}
 
 	protected BiPredicate<ConsumerRecord<?, ?>, Exception> getSkipPredicate(List<ConsumerRecord<?, ?>> records,
@@ -191,7 +218,7 @@ public void clearThreadState() {
 		this.failureTracker.clearThreadState();
 	}
 
-	private static BinaryExceptionClassifier configureDefaultClassifier() {
+	private static ExtendedBinaryExceptionClassifier configureDefaultClassifier() {
 		Map<Class<? extends Throwable>, Boolean> classified = new HashMap<>();
 		classified.put(DeserializationException.class, false);
 		classified.put(MessageConversionException.class, false);

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

@@ -75,7 +75,7 @@ public void testClassifier() {
 		verify(template).sendOffsetsToTransaction(anyMap());
 		verify(template, never()).sendOffsetsToTransaction(anyMap(), any(ConsumerGroupMetadata.class));
 		assertThat(recovered.get()).isSameAs(record1);
-		processor.addNotRetryableException(IllegalStateException.class);
+		processor.addNotRetryableExceptions(IllegalStateException.class);
 		recovered.set(null);
 		recovererShouldFail.set(true);
 		processor.process(records, consumer, illegalState, true, EOSMode.ALPHA);

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

@@ -68,7 +68,7 @@ consumer, mock(MessageListenerContainer.class)))
 		handler.handle(new DeserializationException("intended", null, false, illegalState), records,
 				consumer, mock(MessageListenerContainer.class));
 		assertThat(recovered.get()).isSameAs(record1);
-		handler.addNotRetryableException(IllegalStateException.class);
+		handler.addNotRetryableExceptions(IllegalStateException.class);
 		recovered.set(null);
 		recovererShouldFail.set(true);
 		assertThatExceptionOfType(RuntimeException.class).isThrownBy(() ->