GH-1582: Add GenericErrorHandler.setAckAfterHandle

Resolves #1582

- allow users to use `getContainerProperties().getGenericErrorHandler().setAckAfterHandle(false)`
- add the property to error handlers where it was missing (except logging handlers)

**cherry-pick to 2.5.x**

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -66,4 +66,14 @@ default boolean isAckAfterHandle() {
 		return true;
 	}
 
+	/**
+	 * Set to false to prevent the container from committing the offset of a recovered
+	 * record (when the error handler does not itself throw an exception).
+	 * @param ack false to not commit.
+	 * @since 2.5.6
+	 */
+	default void setAckAfterHandle(boolean ack) {
+		throw new UnsupportedOperationException("This error handler does not support setting this property");
+	}
+
 }

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

@@ -60,6 +60,8 @@ public class RecoveringBatchErrorHandler extends FailedRecordProcessor
 
 	private final SeekToCurrentBatchErrorHandler fallbackHandler = new SeekToCurrentBatchErrorHandler();
 
+	private boolean ackAfterHandle = true;
+
 	/**
 	 * Construct an instance with the default recoverer which simply logs the record after
 	 * {@value SeekUtils#DEFAULT_MAX_FAILURES} (maxFailures) have occurred for a
@@ -102,6 +104,16 @@ public RecoveringBatchErrorHandler(@Nullable BiConsumer<ConsumerRecord<?, ?>, Ex
 		this.fallbackHandler.setBackOff(backOff);
 	}
 
+	@Override
+	public boolean isAckAfterHandle() {
+		return this.ackAfterHandle;
+	}
+
+	@Override
+	public void setAckAfterHandle(boolean ackAfterHandle) {
+		this.ackAfterHandle = ackAfterHandle;
+	}
+
 	@Override
 	public void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container) {

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

@@ -53,6 +53,8 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware
 
 	private final SeekToCurrentBatchErrorHandler seeker = new SeekToCurrentBatchErrorHandler();
 
+	private boolean ackAfterHandle = true;
+
 	/**
 	 * Construct an instance with a default {@link FixedBackOff} (unlimited attempts with
 	 * a 5 second back off).
@@ -80,6 +82,17 @@ public RetryingBatchErrorHandler(BackOff backOff, @Nullable ConsumerRecordRecove
 		};
 	}
 
+	@Override
+	public boolean isAckAfterHandle() {
+		return this.ackAfterHandle;
+	}
+
+	@Override
+	public void setAckAfterHandle(boolean ackAfterHandle) {
+		this.ackAfterHandle = ackAfterHandle;
+	}
+
+
 	@Override
 	public void handle(Exception thrownException, ConsumerRecords<?, ?> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container, Runnable invokeListener) {

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

@@ -96,26 +96,22 @@ public void setCommitRecovered(boolean commitRecovered) { // NOSONAR enhanced ja
 		super.setCommitRecovered(commitRecovered);
 	}
 
-	@Override
-	public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
-			Consumer<?, ?> consumer, MessageListenerContainer container) {
-
-		SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(),
-				getSkipPredicate(records, thrownException), this.logger, getLogLevel());
-	}
-
 	@Override
 	public boolean isAckAfterHandle() {
 		return this.ackAfterHandle;
 	}
 
-	/**
-	 * Set to false to tell the container to NOT commit the offset for a recovered record.
-	 * @param ackAfterHandle false to suppress committing the offset.
-	 * @since 2.3.2
-	 */
+	@Override
 	public void setAckAfterHandle(boolean ackAfterHandle) {
 		this.ackAfterHandle = ackAfterHandle;
 	}
 
+	@Override
+	public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
+			Consumer<?, ?> consumer, MessageListenerContainer container) {
+
+		SeekUtils.seekOrRecover(thrownException, records, consumer, container, isCommitRecovered(),
+				getSkipPredicate(records, thrownException), this.logger, getLogLevel());
+	}
+
 }