GH-1273: Fix offset commit after recovery

Resolves #1273

Previously, offsets were only committed if an error handler "handled" an
error (did not throw an exception), when transactions are being used.

Now, if an error handler successfully recovers, the offset will automatically
be committed.

Since this is a behavior change, a new boolean `ackAfterHandle()` is added to
the `GenericErrorHandler`, default `false` and the `SeekToCurrentErrorHandler`
returns `true`; it is the only framwork handler that can recover from failures.

In a future release, this boolean will be `true` by default.

* Fix default ackAfterHandle=true

* Add docs

Author: garyrussell

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

@@ -56,4 +56,15 @@ default void clearThreadState() {
 		// NOSONAR
 	}
 
+	/**
+	 * Return true if the offset should be committed for a handled error (no exception
+	 * thrown).
+	 * @return true to commit.
+	 * @since 2.3.2
+	 */
+	default boolean isAckAfterHandle() {
+		// TODO: Default true in the next release.
+		return false;
+	}
+
 }

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

@@ -1264,14 +1264,24 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 			}
 			catch (RuntimeException e) {
 				failureTimer(sample);
-				if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) {
+				boolean acked = this.containerProperties.isAckOnError() && !this.autoCommit && producer == null;
+				if (acked) {
 					this.acks.addAll(getHighestOffsetRecords(records));
 				}
 				if (this.batchErrorHandler == null) {
 					throw e;
 				}
 				try {
-					invokeBatchErrorHandler(records, producer, e);
+					invokeBatchErrorHandler(records, e);
+					// unlikely, but possible, that a batch error handler "handles" the error
+					if ((!acked && !this.autoCommit && this.batchErrorHandler.isAckAfterHandle()) || producer != null) {
+						if (!acked) {
+							this.acks.addAll(getHighestOffsetRecords(records));
+						}
+						if (producer != null) {
+							sendOffsetsToTransaction(producer);
+						}
+					}
 				}
 				catch (RuntimeException ee) {
 					this.logger.error(ee, "Error handler threw an exception");
@@ -1376,21 +1386,14 @@ private void doInvokeBatchOnMessage(final ConsumerRecords<K, V> records,
 			}
 		}
 
-		private void invokeBatchErrorHandler(final ConsumerRecords<K, V> records,
-				@SuppressWarnings(RAW_TYPES) @Nullable Producer producer, RuntimeException e) {
-
+		private void invokeBatchErrorHandler(final ConsumerRecords<K, V> records, RuntimeException e) {
 			if (this.batchErrorHandler instanceof ContainerAwareBatchErrorHandler) {
 				this.batchErrorHandler.handle(decorateException(e), records, this.consumer,
 						KafkaMessageListenerContainer.this.container);
 			}
 			else {
 				this.batchErrorHandler.handle(decorateException(e), records, this.consumer);
 			}
-			// if the handler handled the error (no exception), go ahead and commit
-			if (producer != null) {
-				this.acks.addAll(getHighestOffsetRecords(records));
-				sendOffsetsToTransaction(producer);
-			}
 		}
 
 		private void invokeRecordListener(final ConsumerRecords<K, V> records) {
@@ -1536,14 +1539,18 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 			}
 			catch (RuntimeException e) {
 				failureTimer(sample);
-				if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) {
+				boolean acked = this.containerProperties.isAckOnError() && !this.autoCommit && producer == null;
+				if (acked) {
 					ackCurrent(record);
 				}
 				if (this.errorHandler == null) {
 					throw e;
 				}
 				try {
 					invokeErrorHandler(record, producer, iterator, e);
+					if ((!acked && !this.autoCommit && this.errorHandler.isAckAfterHandle()) || producer != null) {
+						ackCurrent(record, producer);
+					}
 				}
 				catch (RuntimeException ee) {
 					this.logger.error(ee, "Error handler threw an exception");
@@ -1629,9 +1636,6 @@ private void invokeErrorHandler(final ConsumerRecord<K, V> record,
 			else {
 				this.errorHandler.handle(decorateException(e), record, this.consumer);
 			}
-			if (producer != null) {
-				ackCurrent(record, producer);
-			}
 		}
 
 		private Exception decorateException(RuntimeException e) {

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

@@ -54,6 +54,8 @@ public class SeekToCurrentErrorHandler extends FailedRecordProcessor implements
 
 	private static final LoggingCommitCallback LOGGING_COMMIT_CALLBACK = new LoggingCommitCallback();
 
+	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
@@ -166,6 +168,7 @@ public void setCommitRecovered(boolean commitRecovered) { // NOSONAR enhanced ja
 	 * @since 2.3
 	 * @deprecated in favor of {@link #setClassifications(Map, boolean)}.
 	 */
+	@Override
 	@Deprecated
 	public void setClassifier(BinaryExceptionClassifier classifier) {
 		Assert.notNull(classifier, "'classifier' + cannot be null");
@@ -205,4 +208,18 @@ public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records
 		}
 	}
 
+	@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
+	 */
+	public void setAckAfterHandle(boolean ackAfterHandle) {
+		this.ackAfterHandle = ackAfterHandle;
+	}
+
 }

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

@@ -99,6 +99,7 @@
 import org.springframework.kafka.test.utils.ContainerTestUtils;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.backoff.FixedBackOff;
 
 /**
  * Tests for the listener container.
@@ -615,6 +616,71 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		container.stop();
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testRecordAckAfterRecoveryMock() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
+		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
+		records.put(new TopicPartition("foo", 0), Arrays.asList(
+				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
+				new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
+		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		given(consumer.poll(any(Duration.class))).willAnswer(i -> {
+			Thread.sleep(50);
+			return consumerRecords;
+		});
+		TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
+				new TopicPartitionOffset("foo", 0) };
+		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+		containerProps.setGroupId("grp");
+		containerProps.setAckMode(AckMode.RECORD);
+		containerProps.setMissingTopicsFatal(false);
+		final CountDownLatch latch = new CountDownLatch(2);
+		MessageListener<Integer, String> messageListener = spy(
+				new MessageListener<Integer, String>() { // Cannot be lambda: Mockito doesn't mock final classes
+
+					@Override
+					public void onMessage(ConsumerRecord<Integer, String> data) {
+						latch.countDown();
+						if (latch.getCount() == 0) {
+							records.clear();
+						}
+						if (data.offset() == 1L) {
+							throw new IllegalStateException();
+						}
+					}
+
+				});
+
+		final CountDownLatch commitLatch = new CountDownLatch(2);
+
+		willAnswer(i -> {
+					commitLatch.countDown();
+					return null;
+				}
+		).given(consumer).commitSync(anyMap(), any());
+
+		containerProps.setMessageListener(messageListener);
+		containerProps.setClientId("clientId");
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		SeekToCurrentErrorHandler errorHandler = spy(new SeekToCurrentErrorHandler(new FixedBackOff(0L, 0)));
+		container.setErrorHandler(errorHandler);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		InOrder inOrder = inOrder(messageListener, consumer, errorHandler);
+		inOrder.verify(consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
+		inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
+		inOrder.verify(consumer).commitSync(anyMap(), any());
+		inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
+		inOrder.verify(errorHandler).handle(any(), any(), any(), any());
+		inOrder.verify(consumer).commitSync(anyMap(), any());
+		container.stop();
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testRecordAckAfterStop() throws Exception {
@@ -1121,6 +1187,66 @@ public void testBatchListenerErrors() throws Exception {
 		logger.info("Stop batch listener errors");
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testBatchListenerAckAfterRecoveryMock() throws Exception {
+		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
+		Consumer<Integer, String> consumer = mock(Consumer.class);
+		given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
+		final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
+		records.put(new TopicPartition("foo", 0), Arrays.asList(
+				new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
+				new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
+		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		given(consumer.poll(any(Duration.class))).willAnswer(i -> {
+			Thread.sleep(50);
+			return consumerRecords;
+		});
+		TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
+				new TopicPartitionOffset("foo", 0) };
+		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+		containerProps.setGroupId("grp");
+		containerProps.setMissingTopicsFatal(false);
+		final CountDownLatch latch = new CountDownLatch(1);
+		BatchMessageListener<Integer, String> messageListener = spy(
+				new BatchMessageListener<Integer, String>() { // Cannot be lambda: Mockito doesn't mock final classes
+
+					@Override
+					public void onMessage(List<ConsumerRecord<Integer, String>> data) {
+						latch.countDown();
+						throw new IllegalStateException();
+					}
+
+
+				});
+
+		final CountDownLatch commitLatch = new CountDownLatch(1);
+
+		willAnswer(i -> {
+					commitLatch.countDown();
+					records.clear();
+					return null;
+				}
+		).given(consumer).commitSync(anyMap(), any());
+
+		containerProps.setMessageListener(messageListener);
+		containerProps.setClientId("clientId");
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		BatchErrorHandler errorHandler = mock(BatchErrorHandler.class);
+		given(errorHandler.isAckAfterHandle()).willReturn(true);
+		container.setBatchErrorHandler(errorHandler);
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		InOrder inOrder = inOrder(messageListener, consumer, errorHandler);
+		inOrder.verify(consumer).poll(Duration.ofMillis(ContainerProperties.DEFAULT_POLL_TIMEOUT));
+		inOrder.verify(messageListener).onMessage(any());
+		inOrder.verify(errorHandler).handle(any(), any(), any());
+		inOrder.verify(consumer).commitSync(anyMap(), any());
+		container.stop();
+	}
+
 	@Test
 	public void testSeek() throws Exception {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test11", "false", embeddedKafka);

src/reference/asciidoc/kafka.adoc

@@ -1736,11 +1736,9 @@ IMPORTANT: The `FilteringBatchMessageListenerAdapter` is ignored if your `@Kafka
 [[retrying-deliveries]]
 ===== Retrying Deliveries
 
-If your listener throws an exception, the default behavior is to invoke the `ErrorHandler`, if configured, or logged otherwise.
-
-NOTE: Two error handler interfaces (`ErrorHandler` and `BatchErrorHandler`) are provided.
-You must configure the appropriate type to match the <<message-listeners,message listener>>.
+If your listener throws an exception, the default behavior is to invoke the <<error-handlers>>, if configured, or logged otherwise.
 
+NOTE:
 To retry deliveries, a convenient listener adapter `RetryingMessageListenerAdapter`  is provided.
 
 You can configure it with a `RetryTemplate` and `RecoveryCallback<Void>` - see the https://github.com/spring-projects/spring-retry[spring-retry] project for information about these components.
@@ -3400,8 +3398,22 @@ This resets each topic/partition in the batch to the lowest offset in the batch.
 
 NOTE: The preceding two examples are simplistic implementations, and you would probably want more checking in the error handler.
 
+[[error-handlers]]
 ===== Container Error Handlers
 
+Two error handler interfaces (`ErrorHandler` and `BatchErrorHandler`) are provided.
+You must configure the appropriate type to match the <<message-listeners,message listener>>.
+
+By default, errors are simply logged when transactions are not being used.
+When transactions are being used, no error handlers are configured, by default, so that the exception will roll back the transaction.
+If you provide a custom error handler when using transactions, it must throw an exception if you want the transaction rolled back.
+
+Starting with version 2.3.2, these interfaces have a default method `isAckAfterHandle()` which is called by the container to determine whether the offset(s) should be committed if the error handler returns without throwing an exception.
+This returns false by default, for backwards compatibility.
+In most cases, however, we expect that the offset should be committed.
+For example, the <<seek-to-current, `SeekToCurrentErrorHandler`>> returns `true` if a record is recovered (after any retries, if so configured).
+In a future release, we expect to change this default to `true`.
+
 You can specify a global error handler to be used for all listeners in the container factory.
 The following example shows how to do so:
 
@@ -3441,6 +3453,8 @@ public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer,
 
 By default, if an annotated listener method throws an exception, it is thrown to the container, and the message is handled according to the container configuration.
 
+If you are using Spring Boot, you simply need to add the error handler as a `@Bean` and boot will add it to the auto-configured factory.
+
 ===== Consumer-Aware Container Error Handlers
 
 The container-level error handlers (`ErrorHandler` and `BatchErrorHandler`) have sub-interfaces called `ConsumerAwareErrorHandler` and `ConsumerAwareBatchErrorHandler`.
@@ -3593,6 +3607,9 @@ Again, the maximum delay must be less than the `max.poll.interval.ms` consumer p
 
 IMPORTANT: If the recoverer fails (throws an exception), the record will be included in the seeks and recovery will be attempted again during the next delivery.
 
+Starting with version 2.3.2, after a record has been recovered, its offset will be committed (if one of the container `AckMode` s is configured).
+To revert to the previous behavior, set the error handler's `ackAfterHandle` property to false.
+
 ===== Container Stopping Error Handlers
 
 The `ContainerStoppingErrorHandler` (used with record listeners) stops the container if the listener throws an exception.

src/reference/asciidoc/whats-new.adoc

@@ -70,6 +70,8 @@ The `SeekToCurrentErrorHandler` now treats certain exceptions as fatal and disab
 
 The `SeekToCurrentErrorHandler` and `SeekToCurrentBatchErrorHandler` can now be configured to apply a `BackOff` (thread sleep) between delivery attempts.
 
+Starting with version 2.3.2, recovered records' offsets will be committed when the error handler returns after recovering a failed record.
+
 See <<seek-to-current>> for more information.
 
 The `DeadLetterPublishingRecoverer`, when used in conjunction with an `ErrorHandlingDeserializer2`, now sets the payload of the message sent to the dead-letter topic, to the original value that could not be deserialized.