GH-1522: Commit recovered record with MANUAL Acks

Resolves #1522

If the `errorHandler.isAckAfterHandle()` is true and the handler exits
normally, add the offset to the pending `acks`, even when the `AckMode`
is `MANUAL`.

Author: garyrussell

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

@@ -578,6 +578,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private boolean producerPerConsumerPartition;
 
+		private boolean commitRecovered;
+
 		private volatile boolean consumerPaused;
 
 		private volatile Thread consumerThread;
@@ -1794,7 +1796,13 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 					invokeErrorHandler(record, iterator, e);
 					if ((!acked && !this.autoCommit && this.errorHandler.isAckAfterHandle())
 							|| this.producer != null) {
+						if (this.isManualAck) {
+							this.commitRecovered = true;
+						}
 						ackCurrent(record);
+						if (this.isManualAck) {
+							this.commitRecovered = false;
+						}
 					}
 				}
 				catch (KafkaException ke) {
@@ -1932,7 +1940,8 @@ public void ackCurrent(final ConsumerRecord<K, V> record) {
 					this.acks.add(record);
 				}
 			}
-			else if (this.producer != null || (!this.isAnyManualAck && !this.autoCommit)) {
+			else if (this.producer != null
+					|| ((!this.isAnyManualAck || this.commitRecovered) && !this.autoCommit)) {
 				this.acks.add(record);
 			}
 			if (this.producer != null) {

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

@@ -174,7 +174,7 @@ public static void seekOrRecover(Exception thrownException, List<ConsumerRecord<
 				}
 			}
 			else {
-				logger.warn(() -> "'commitRecovered' ignored, container AckMode must be MANUAL_IMMEDIATE, not "
+				logger.debug(() -> "'commitRecovered' ignored, container AckMode must be MANUAL_IMMEDIATE, not "
 						+ container.getContainerProperties().getAckMode());
 			}
 		}

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

@@ -3081,6 +3081,51 @@ private void testCommitRetriesGuts(boolean sync) throws Exception {
 		}
 	}
 
+	@Test
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	void commitAfterHandleManual() throws InterruptedException {
+		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);
+		Map<String, Object> cfProps = new HashMap<>();
+		cfProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 45000); // wins
+		given(cf.getConfigurationProperties()).willReturn(cfProps);
+		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")));
+		ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
+		ConsumerRecords<Integer, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
+		AtomicBoolean first = new AtomicBoolean(true);
+		given(consumer.poll(any(Duration.class))).willAnswer(i -> {
+			Thread.sleep(50);
+			return first.getAndSet(false) ? consumerRecords : emptyRecords;
+		});
+		TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
+				new TopicPartitionOffset("foo", 0) };
+		ContainerProperties containerProps = new ContainerProperties(topicPartition);
+		containerProps.setAckMode(AckMode.MANUAL);
+		containerProps.setGroupId("grp");
+		containerProps.setClientId("clientId");
+		containerProps.setIdleEventInterval(100L);
+		containerProps.setMessageListener((MessageListener) r -> {
+			throw new RuntimeException("test");
+		});
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		AtomicBoolean recovered = new AtomicBoolean();
+		CountDownLatch latch = new CountDownLatch(1);
+		container.setErrorHandler(new SeekToCurrentErrorHandler((rec, ex) -> {
+					recovered.set(true);
+					latch.countDown();
+				},
+				new FixedBackOff(0, 0)));
+		container.start();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		container.stop();
+		assertThat(recovered.get()).isTrue();
+		verify(consumer).commitSync(any(), any());
+	}
+
 	private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
 		Consumer<?, ?> consumer =
 				KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class);