spring-projectsGH-2195: DefaultErrorHandlerImprovements

Resolves spring-projects#2195

Add an option to avoid seeks after handling exceptions.

Instead, pause the consumer for one `poll()` and use the remaining records as the
result of that poll.

New methods on `CommonErrorHandler` - `handleOne` for record listeners, returning
a boolean to indicate whether the record was recovered and should not be redelivered.

`handlaBatchAndReturnRemaining` for batch listeners, returning either the complete
set or a subset, e.g. when the `DEH` receives a `BatchListenerExecutionFailedException`
and commits a partial batch.

Also includes the classifier refactoring discussed here
spring-projects#2185 (comment)

The new logic is disabled by default, we can consider enabling it in 3.0 and remove
the deprecations.

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -41,13 +41,25 @@ public interface CommonErrorHandler extends DeliveryAttemptAware {
 	 * When true (default), all remaining records including the failed record are passed
 	 * to the error handler.
 	 * @return false to receive only the failed record.
+	 * @deprecated in favor of {@link #seeksAfterHandling()}.
 	 * @see #handleRecord(Exception, ConsumerRecord, Consumer, MessageListenerContainer)
 	 * @see #handleRemaining(Exception, List, Consumer, MessageListenerContainer)
 	 */
+	@Deprecated
 	default boolean remainingRecords() {
 		return false;
 	}
 
+	/**
+	 * Return true if this error handle performs seeks on the failed record and remaining
+	 * records (or just the remaining records after a failed record is recovered).
+	 * @return true if the next poll should fetch records.
+	 */
+	@SuppressWarnings("deprecation")
+	default boolean seeksAfterHandling() {
+		return remainingRecords();
+	}
+
 	/**
 	 * Return true if this error handler supports delivery attempts headers.
 	 * @return true if capable.
@@ -79,14 +91,41 @@ default void handleOtherException(Exception thrownException, Consumer<?, ?> cons
 	 * @param record the record.
 	 * @param consumer the consumer.
 	 * @param container the container.
+	 * @deprecated in favor of
+	 * {@link #handleOne(Exception, ConsumerRecord, Consumer, MessageListenerContainer)}.
 	 * @see #remainingRecords()
 	 */
+	@Deprecated
 	default void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
 		LogFactory.getLog(getClass()).error("'handleRecord' is not implemented by this handler", thrownException);
 	}
 
+	/**
+	 * Handle the exception for a record listener when {@link #remainingRecords()} returns
+	 * false. Use this to handle just the single failed record.
+	 * @param thrownException the exception.
+	 * @param record the record.
+	 * @param consumer the consumer.
+	 * @param container the container.
+	 * @return true if the error was "handled" or false if not and the container will
+	 * re-submit the record to the listener.
+	 * @see #remainingRecords()
+	 */
+	@SuppressWarnings("deprecation")
+	default boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+
+		try {
+			handleRecord(thrownException, record, consumer, container);
+			return true;
+		}
+		catch (Exception ex) {
+			return false;
+		}
+	}
+
 	/**
 	 * Handle the exception for a record listener when {@link #remainingRecords()} returns
 	 * true. The failed record and all the remaining records from the poll are passed in.
@@ -120,6 +159,28 @@ default void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data,
 		LogFactory.getLog(getClass()).error("'handleBatch' is not implemented by this handler", thrownException);
 	}
 
+	/**
+	 * Handle the exception for a batch listener. The complete {@link ConsumerRecords}
+	 * from the poll is supplied. Return the members of the batch that should be re-sent to
+	 * the listener. The returned records MUST be in the same order as the original records.
+	 * @param thrownException the exception.
+	 * @param data the consumer records.
+	 * @param consumer the consumer.
+	 * @param container the container.
+	 * @param invokeListener a callback to re-invoke the listener.
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 * @return the consumer records, or a subset.
+	 * @since 2.8.5
+	 */
+	default <K, V> ConsumerRecords<K, V> handleBatchAndReturnRemaining(Exception thrownException,
+			ConsumerRecords<?, ?> data, Consumer<?, ?> consumer, MessageListenerContainer container,
+			Runnable invokeListener) {
+
+		handleBatch(thrownException, data, consumer, container, invokeListener);
+		return ConsumerRecords.empty();
+	}
+
 	@Override
 	default int deliveryAttempt(TopicPartitionOffset topicPartitionOffset) {
 		return 0;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -73,10 +73,10 @@ public void handleOtherException(Exception thrownException, Consumer<?, ?> consu
 	}
 
 	@Override
-	public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
+	public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {
 
-		this.recordErrorHandler.handleRecord(thrownException, record, consumer, container);
+		return this.recordErrorHandler.handleOne(thrownException, record, consumer, container);
 	}
 
 	@Override

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

@@ -133,8 +133,9 @@ private void checkConfig() {
 	public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer,
 			@Nullable MessageListenerContainer container, Exception exception, boolean recoverable, EOSMode eosMode) {
 
-		if (SeekUtils.doSeeks(((List) records), consumer, exception, recoverable,
-				getRecoveryStrategy((List) records, exception), container, this.logger)
+		List records2 = records;
+		if (SeekUtils.doSeeks(records2, consumer, exception, recoverable,
+				getFailureTracker()::recovered, container, this.logger)
 					&& isCommitRecovered() && this.kafkaTemplate.isTransactional()) {
 			ConsumerRecord<K, V> skipped = records.get(0);
 			this.kafkaTemplate.sendOffsetsToTransaction(

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

@@ -118,20 +118,37 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 
 	@Override
 	public boolean remainingRecords() {
-		return true;
+		return isSeekAfterError();
+	}
+
+	@Override
+	public boolean seeksAfterHandling() {
+		return remainingRecords();
 	}
 
 	@Override
 	public boolean deliveryAttemptHeader() {
 		return true;
 	}
 
+	@Override
+	public boolean handleOne(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+
+		try {
+			return getFailureTracker().recovered(record, thrownException, container, consumer);
+		}
+		catch (Exception ex) {
+			return false;
+		}
+	}
+
 	@Override
 	public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container) {
 
 		SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(), // NOSONAR
-				getRecoveryStrategy(records, consumer, thrownException), this.logger, getLogLevel());
+				getFailureTracker()::recovered, this.logger, getLogLevel());
 	}
 
 	@Override
@@ -141,6 +158,14 @@ public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, C
 		doHandle(thrownException, data, consumer, container, invokeListener);
 	}
 
+	@Override
+	public <K, V> ConsumerRecords<K, V> handleBatchAndReturnRemaining(Exception thrownException,
+			ConsumerRecords<?, ?> data, Consumer<?, ?> consumer, MessageListenerContainer container,
+			Runnable invokeListener) {
+
+		return handle(thrownException, data, consumer, container, invokeListener);
+	}
+
 	@Override
 	public void handleOtherException(Exception thrownException, Consumer<?, ?> consumer,
 			MessageListenerContainer container, boolean batchListener) {

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

@@ -72,6 +72,12 @@ public FailedBatchProcessor(@Nullable BiConsumer<ConsumerRecord<?, ?>, Exception
 	protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container, Runnable invokeListener) {
 
+		handle(thrownException, data, consumer, container, invokeListener);
+	}
+
+	protected <K, V> ConsumerRecords handle(Exception thrownException, ConsumerRecords<?, ?> data,
+			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {
+
 		BatchListenerFailedException batchListenerFailedException = getBatchListenerFailedException(thrownException);
 		if (batchListenerFailedException == null) {
 			this.logger.debug(thrownException, "Expected a BatchListenerFailedException; re-seeking batch");
@@ -87,9 +93,10 @@ protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> data, C
 				this.fallbackBatchHandler.handleBatch(thrownException, data, consumer, container, invokeListener);
 			}
 			else {
-				seekOrRecover(thrownException, data, consumer, container, index);
+				return seekOrRecover(thrownException, data, consumer, container, index);
 			}
 		}
+		return ConsumerRecords.empty();
 	}
 
 	private int findIndex(ConsumerRecords<?, ?> data, ConsumerRecord<?, ?> record) {
@@ -109,10 +116,12 @@ private int findIndex(ConsumerRecords<?, ?> data, ConsumerRecord<?, ?> record) {
 		return i;
 	}
 
-	private void seekOrRecover(Exception thrownException, @Nullable ConsumerRecords<?, ?> data, Consumer<?, ?> consumer, MessageListenerContainer container, int indexArg) {
+	@SuppressWarnings("unchecked")
+	private <K, V> ConsumerRecords<K, V> seekOrRecover(Exception thrownException, @Nullable ConsumerRecords<?, ?> data,
+			Consumer<?, ?> consumer, MessageListenerContainer container, int indexArg) {
 
 		if (data == null) {
-			return;
+			return ConsumerRecords.empty();
 		}
 		Iterator<?> iterator = data.iterator();
 		List<ConsumerRecord<?, ?>> toCommit = new ArrayList<>();
@@ -133,15 +142,37 @@ private void seekOrRecover(Exception thrownException, @Nullable ConsumerRecords<
 		if (offsets.size() > 0) {
 			commit(consumer, container, offsets);
 		}
-		if (remaining.size() > 0) {
-			SeekUtils.seekOrRecover(thrownException, remaining, consumer, container, false,
-				getRecoveryStrategy(remaining, thrownException), this.logger, getLogLevel());
-			ConsumerRecord<?, ?> recovered = remaining.get(0);
-			commit(consumer, container,
-					Collections.singletonMap(new TopicPartition(recovered.topic(), recovered.partition()),
-							new OffsetAndMetadata(recovered.offset() + 1)));
-			if (remaining.size() > 1) {
-				throw new KafkaException("Seek to current after exception", getLogLevel(), thrownException);
+		if (isSeekAfterError()) {
+			if (remaining.size() > 0) {
+				SeekUtils.seekOrRecover(thrownException, remaining, consumer, container, false,
+					getFailureTracker()::recovered, this.logger, getLogLevel());
+				ConsumerRecord<?, ?> recovered = remaining.get(0);
+				commit(consumer, container,
+						Collections.singletonMap(new TopicPartition(recovered.topic(), recovered.partition()),
+								new OffsetAndMetadata(recovered.offset() + 1)));
+				if (remaining.size() > 1) {
+					throw new KafkaException("Seek to current after exception", getLogLevel(), thrownException);
+				}
+			}
+			return ConsumerRecords.empty();
+		}
+		else {
+			if (indexArg == 0) {
+				return (ConsumerRecords<K, V>) data; // first record just rerun the whole thing
+			}
+			else {
+				try {
+					if (getFailureTracker().recovered(remaining.get(0), thrownException, container,
+							consumer)) {
+						remaining.remove(0);
+					}
+				}
+				catch (Exception e) {
+				}
+				Map<TopicPartition, List<ConsumerRecord<K, V>>> remains = new HashMap<>();
+				remaining.forEach(rec -> remains.computeIfAbsent(new TopicPartition(rec.topic(), rec.partition()),
+						tp -> new ArrayList<ConsumerRecord<K, V>>()).add((ConsumerRecord<K, V>) rec));
+				return new ConsumerRecords<>(remains);
 			}
 		}
 	}