GH-1287: Detect mis-configured deserialization

Resolves #1287

Throw an `IllegalStateException` in the `SeekToCurrentErrorHandler` if
the root exception is a `SerializationException`.

Handling of deserializatin problms requires the configuration of an
`ErrorHandlingDeserializer2`.

**cherry-pick to 2.2.x, 2.1.x**

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

@@ -26,6 +26,7 @@
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.SerializationException;
 
 import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.kafka.KafkaException;
@@ -180,6 +181,12 @@ public void setClassifier(BinaryExceptionClassifier classifier) {
 	public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container) {
 
+		if (thrownException instanceof SerializationException) {
+			throw new IllegalStateException("This error handler cannot process 'SerializationException's directly, "
+					+ "please consider configuring an 'ErrorHandlingDeserializer2' in the value and/or key "
+					+ "deserializer", thrownException);
+		}
+
 		if (!SeekUtils.doSeeks(records, consumer, thrownException, true, getSkipPredicate(records, thrownException),
 				LOGGER)) {
 			throw new KafkaException("Seek to current after exception", thrownException);

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

@@ -18,6 +18,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -30,6 +31,7 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.SerializationException;
 import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;
 
@@ -93,4 +95,13 @@ public void testDeprecatedCtor() {
 		assertThat(KafkaTestUtils.getPropertyValue(handler, "failureTracker.backOff.maxAttempts"))
 				.isEqualTo(9L);
 	}
+
+	@Test
+	void testSerializationException() {
+		SeekToCurrentErrorHandler handler = new SeekToCurrentErrorHandler();
+		SerializationException thrownException = new SerializationException();
+		assertThatIllegalStateException().isThrownBy(() -> handler.handle(thrownException, null, null, null))
+				.withCause(thrownException);
+	}
+
 }