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 d4c2081b..f6e35ad4 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 @@ -37,6 +37,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; import org.springframework.util.backoff.BackOff; import org.springframework.util.backoff.FixedBackOff; @@ -181,10 +182,17 @@ 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 (ObjectUtils.isEmpty(records)) { + 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); + } + else { + throw new IllegalStateException("This error handler cannot process '" + + thrownException.getClass().getName() + + "'s; no record information is available", thrownException); + } } if (!SeekUtils.doSeeks(records, consumer, thrownException, true, getSkipPredicate(records, 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 d89a35d1..656ba593 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 @@ -24,6 +24,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; @@ -104,4 +105,13 @@ public class SeekToCurrentErrorHandlerTests { .withCause(thrownException); } + @Test + void testNotRetryableWithNoRecords() { + SeekToCurrentErrorHandler handler = new SeekToCurrentErrorHandler(); + ClassCastException thrownException = new ClassCastException(); + assertThatIllegalStateException().isThrownBy( + () -> handler.handle(thrownException, Collections.emptyList(), null, null)) + .withCause(thrownException); + } + }