GH-983: Seek remaining records

Resolves #983

When skipping non-retryable exceptions, we still need to seek the
remaining topic/partitions.

Author: garyrussell

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

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.BiConsumer;
+import java.util.function.BiPredicate;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -53,6 +54,8 @@
  */
 public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler {
 
+	private static final BiPredicate<ConsumerRecord<?, ?>, Exception> ALWAYS_SKIP_PREDICATE = (r, e) -> true;
+
 	protected static final Log LOGGER = LogFactory.getLog(SeekToCurrentErrorHandler.class); // NOSONAR visibility
 
 	private static final LoggingCommitCallback LOGGING_COMMIT_CALLBACK = new LoggingCommitCallback();
@@ -101,9 +104,7 @@ public SeekToCurrentErrorHandler(BiConsumer<ConsumerRecord<?, ?>, Exception> rec
 	 * @param maxFailures the maxFailures; a negative value is treated as infinity.
 	 * @since 2.2
 	 */
-	public SeekToCurrentErrorHandler(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer,
-			int maxFailures) {
-
+	public SeekToCurrentErrorHandler(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception> recoverer, int maxFailures) {
 		this.failureTracker = new FailedRecordTracker(recoverer, maxFailures, LOGGER);
 		this.classifier = configureDefaultClassifier();
 	}
@@ -213,10 +214,8 @@ public boolean removeNotRetryableException(Class<? extends Exception> exceptionT
 	public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container) {
 
-		if (!this.classifier.classify(thrownException)) {
-			this.failureTracker.getRecoverer().accept(records.get(0), thrownException);
-		}
-		else if (!SeekUtils.doSeeks(records, consumer, thrownException, true, this.failureTracker::skip, LOGGER)) {
+		if (!SeekUtils.doSeeks(records, consumer, thrownException, true, getSkipPredicate(records, thrownException),
+				LOGGER)) {
 			throw new KafkaException("Seek to current after exception", thrownException);
 		}
 		if (this.commitRecovered) {
@@ -242,6 +241,16 @@ else if (!SeekUtils.doSeeks(records, consumer, thrownException, true, this.failu
 		}
 	}
 
+	private BiPredicate<ConsumerRecord<?, ?>, Exception> getSkipPredicate(List<ConsumerRecord<?, ?>> records, Exception thrownException) {
+		if (this.classifier.classify(thrownException)) {
+			return this.failureTracker::skip;
+		}
+		else {
+			this.failureTracker.getRecoverer().accept(records.get(0), thrownException);
+			return ALWAYS_SKIP_PREDICATE;
+		}
+	}
+
 	@Override
 	public void clearThreadState() {
 		this.failureTracker.clearThreadState();

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

@@ -18,15 +18,19 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
 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.junit.jupiter.api.Test;
+import org.mockito.InOrder;
 
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.serializer.DeserializationException;
@@ -42,19 +46,25 @@ public class SeekToCurrentErrorHandlerTests {
 	public void testClassifier() {
 		AtomicReference<ConsumerRecord<?, ?>> recovered = new AtomicReference<>();
 		SeekToCurrentErrorHandler handler = new SeekToCurrentErrorHandler((r, t) -> recovered.set(r));
-		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
-		List<ConsumerRecord<?, ?>> records = Collections.singletonList(record);
+		ConsumerRecord<String, String> record1 = new ConsumerRecord<>("foo", 0, 0L, "foo", "bar");
+		ConsumerRecord<String, String> record2 = new ConsumerRecord<>("foo", 0, 1L, "foo", "bar");
+		List<ConsumerRecord<?, ?>> records = Arrays.asList(record1, record2);
 		IllegalStateException illegalState = new IllegalStateException();
+		Consumer<?, ?> consumer = mock(Consumer.class);
 		assertThatExceptionOfType(KafkaException.class).isThrownBy(() -> handler.handle(illegalState, records,
-					mock(Consumer.class), mock(MessageListenerContainer.class)))
+					consumer, mock(MessageListenerContainer.class)))
 				.withCause(illegalState);
 		handler.handle(new DeserializationException("intended", null, false, illegalState), records,
-				mock(Consumer.class), mock(MessageListenerContainer.class));
-		assertThat(recovered.get()).isSameAs(record);
+				consumer, mock(MessageListenerContainer.class));
+		assertThat(recovered.get()).isSameAs(record1);
 		handler.addNotRetryableException(IllegalStateException.class);
 		recovered.set(null);
-		handler.handle(illegalState, records, mock(Consumer.class), mock(MessageListenerContainer.class));
-		assertThat(recovered.get()).isSameAs(record);
+		handler.handle(illegalState, records, 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", 0), 1L); // 2x recovered seek next
+		inOrder.verifyNoMoreInteractions();
 	}
 
 }