GH-1289: STCEH - Fix IndexOutOfBoundsException

Resolves https://github.com/spring-projects/spring-kafka/issues/1289

Fix IOOBE If a not-retryable exception was thrown outside of record handling
(when no records are passed to the error handler).

Throw an illegal state exception if this condition occurs.

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<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 (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),

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);
+	}
+
 }