From c3ebc734f9dcd26d140dce5718c27574c432270b Mon Sep 17 00:00:00 2001
From: Kwangyong Kim <banana.yong@gmail.com>
Date: Mon, 17 Aug 2020 02:12:45 +0900
Subject: [PATCH] Add ConversionException in classifier to skip

- ConversionException should be considered as fatal like MessageConversionException
- Add ConversionException in FailedRecordProcessor.classifier to skip retrying
- edit reference document
---
 .../kafka/listener/FailedRecordProcessor.java              | 5 +++++
 .../kafka/listener/SeekToCurrentErrorHandlerTests.java     | 7 ++++++-
 src/reference/asciidoc/kafka.adoc                          | 2 ++
 3 files changed, 13 insertions(+), 1 deletion(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
index 4f46260485..0d0fb25879 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/FailedRecordProcessor.java
@@ -29,6 +29,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;
@@ -141,6 +142,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>
@@ -163,6 +165,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>
@@ -191,6 +194,7 @@ public final void addNotRetryableExceptions(Class<? extends Exception>... except
 	 * <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>
@@ -234,6 +238,7 @@ private static ExtendedBinaryExceptionClassifier 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);
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentErrorHandlerTests.java
index 04994f6968..60e127e67e 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentErrorHandlerTests.java
+++ b/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.addNotRetryableExceptions(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();
diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc
index 6472f7874a..04103b1ddc 100644
--- a/src/reference/asciidoc/kafka.adoc
+++ b/src/reference/asciidoc/kafka.adoc
@@ -4611,6 +4611,7 @@ The exceptions that are considered fatal, by default, are:
 
 * `DeserializationException`
 * `MessageConversionException`
+* `ConversionException`
 * `MethodArgumentResolutionException`
 * `NoSuchMethodException`
 * `ClassCastException`
@@ -4863,6 +4864,7 @@ The exceptions that are considered fatal, by default, are:
 
 * `DeserializationException`
 * `MessageConversionException`
+* `ConversionException`
 * `MethodArgumentResolutionException`
 * `NoSuchMethodException`
 * `ClassCastException`