From 30dca499f7bec98a0872c52e800da44f92962619 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 29 Oct 2019 16:31:45 -0400 Subject: [PATCH] GH-1287: Detect mis-configured deserialization Resolves https://github.com/spring-projects/spring-kafka/issues/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** --- .../kafka/listener/SeekToCurrentErrorHandler.java | 7 +++++++ .../listener/SeekToCurrentErrorHandlerTests.java | 11 +++++++++++ 2 files changed, 18 insertions(+) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java index 5ea54d5b..d4c2081b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java @@ -26,6 +26,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; 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 class SeekToCurrentErrorHandler extends FailedRecordProcessor implements public void handle(Exception thrownException, List> 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); 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 9cf1ad6a..d89a35d1 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 @@ -18,6 +18,7 @@ package org.springframework.kafka.listener; 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 java.util.concurrent.atomic.AtomicReference; 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 class SeekToCurrentErrorHandlerTests { 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); + } + }