Add ConversionException in classifier to skip

Resolves #1566

- ConversionException should be considered as fatal like MessageConversionException
- Add ConversionException in FailedRecordProcessor.classifier to skip retrying
- edit reference document

Author: bananayong

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

@@ -28,6 +28,7 @@
 import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.TopicPartitionOffset;
+import org.springframework.kafka.support.converter.ConversionException;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.converter.MessageConversionException;
@@ -128,6 +129,7 @@ public int deliveryAttempt(TopicPartitionOffset topicPartitionOffset) {
 	 * <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>
@@ -150,6 +152,7 @@ public void addNotRetryableException(Class<? extends Exception> exceptionType) {
 	 * <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>
@@ -195,6 +198,7 @@ private static BinaryExceptionClassifier 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);

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

@@ -37,6 +37,7 @@
 import org.mockito.InOrder;
 
 import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.support.converter.ConversionException;
 import org.springframework.kafka.support.serializer.DeserializationException;
 
 /**
@@ -68,6 +69,10 @@ consumer, mock(MessageListenerContainer.class)))
 		handler.handle(new DeserializationException("intended", null, false, illegalState), records,
 				consumer, mock(MessageListenerContainer.class));
 		assertThat(recovered.get()).isSameAs(record1);
+		recovered.set(null);
+		handler.handle(new ConversionException("intended", null), records,
+				consumer, mock(MessageListenerContainer.class));
+		assertThat(recovered.get()).isSameAs(record1);
 		handler.addNotRetryableException(IllegalStateException.class);
 		recovered.set(null);
 		recovererShouldFail.set(true);
@@ -77,7 +82,7 @@ consumer, mock(MessageListenerContainer.class)))
 		assertThat(recovered.get()).isSameAs(record1);
 		InOrder inOrder = inOrder(consumer);
 		inOrder.verify(consumer).seek(new TopicPartition("foo", 0), 0L); // not recovered so seek
-		inOrder.verify(consumer, times(2)).seek(new TopicPartition("foo", 1), 1L);
+		inOrder.verify(consumer, times(3)).seek(new TopicPartition("foo", 1), 1L);
 		inOrder.verify(consumer).seek(new TopicPartition("foo", 0), 0L); // recovery failed
 		inOrder.verify(consumer, times(2)).seek(new TopicPartition("foo", 1), 1L);
 		inOrder.verifyNoMoreInteractions();

src/reference/asciidoc/kafka.adoc

@@ -4600,6 +4600,7 @@ The exceptions that are considered fatal, by default, are:
 
 * `DeserializationException`
 * `MessageConversionException`
+* `ConversionException`
 * `MethodArgumentResolutionException`
 * `NoSuchMethodException`
 * `ClassCastException`
@@ -4819,6 +4820,7 @@ The exceptions that are considered fatal, by default, are:
 
 * `DeserializationException`
 * `MessageConversionException`
+* `ConversionException`
 * `MethodArgumentResolutionException`
 * `NoSuchMethodException`
 * `ClassCastException`