Fix manual acks with transactions

- send the offset to the transaction instead of committing via the consumer.

Author: garyrussell

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

@@ -361,6 +361,8 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private boolean taskSchedulerExplicitlySet;
 
+		private Producer<?, ?> producer;
+
 		private volatile long lastPoll = System.currentTimeMillis();
 
 		@SuppressWarnings("unchecked")
@@ -776,7 +778,10 @@ private void ackImmediate(ConsumerRecord<K, V> record) {
 			if (ListenerConsumer.this.logger.isDebugEnabled()) {
 				ListenerConsumer.this.logger.debug("Committing: " + commits);
 			}
-			if (this.containerProperties.isSyncCommits()) {
+			if (this.producer != null) {
+				this.producer.sendOffsetsToTransaction(commits, this.consumerGroupId);
+			}
+			else if (this.containerProperties.isSyncCommits()) {
 				ListenerConsumer.this.consumer.commitSync(commits);
 			}
 			else {
@@ -821,7 +826,9 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 						Producer producer = null;
 						if (ListenerConsumer.this.kafkaTxManager != null) {
 							producer = ((KafkaResourceHolder) TransactionSynchronizationManager
-									.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())).getProducer();
+									.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
+										.getProducer(); // NOSONAR nullable
+							ListenerConsumer.this.producer = producer;
 						}
 						RuntimeException aborted = doInvokeBatchListener(records, recordList, producer);
 						if (aborted != null) {
@@ -931,7 +938,9 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
 							Producer producer = null;
 							if (ListenerConsumer.this.kafkaTxManager != null) {
 								producer = ((KafkaResourceHolder) TransactionSynchronizationManager
-										.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory())).getProducer();
+										.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
+												.getProducer(); // NOSONAR
+								ListenerConsumer.this.producer = producer;
 							}
 							RuntimeException aborted = doInvokeRecordListener(record, producer);
 							if (aborted != null) {
@@ -987,7 +996,9 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 				else {
 					this.listener.onMessage(record);
 				}
-				ackCurrent(record, producer);
+				if (!this.isManualImmediateAck) {
+					ackCurrent(record, producer);
+				}
 			}
 			catch (RuntimeException e) {
 				if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2018 the original author or authors.
+ * Copyright 2017-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.
@@ -67,7 +67,9 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.core.ProducerFactoryUtils;
+import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
 import org.springframework.kafka.listener.config.ContainerProperties;
+import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
@@ -98,21 +100,28 @@ public class TransactionalContainerTests {
 
 	@Test
 	public void testConsumeAndProduceTransactionKTM() throws Exception {
-		testConsumeAndProduceTransactionGuts(false, false);
+		testConsumeAndProduceTransactionGuts(false, false, AckMode.RECORD);
 	}
 
 	@Test
 	public void testConsumeAndProduceTransactionKCTM() throws Exception {
-		testConsumeAndProduceTransactionGuts(true, false);
+		testConsumeAndProduceTransactionGuts(true, false, AckMode.RECORD);
 	}
 
 	@Test
 	public void testConsumeAndProduceTransactionHandleError() throws Exception {
-		testConsumeAndProduceTransactionGuts(false, true);
+		testConsumeAndProduceTransactionGuts(false, true, AckMode.RECORD);
+	}
+
+	@Test
+	public void testConsumeAndProduceTransactionKTMManual() throws Exception {
+		testConsumeAndProduceTransactionGuts(false, false, AckMode.MANUAL_IMMEDIATE);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
-	private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handleError) throws Exception {
+	private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handleError, AckMode ackMode)
+			throws Exception {
+
 		Consumer consumer = mock(Consumer.class);
 		final TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
@@ -122,14 +131,15 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		}).given(consumer).subscribe(any(Collection.class), any(ConsumerRebalanceListener.class));
 		ConsumerRecords records = new ConsumerRecords(Collections.singletonMap(topicPartition,
 				Collections.singletonList(new ConsumerRecord<>("foo", 0, 0, "key", "value"))));
+		ConsumerRecords empty = new ConsumerRecords(Collections.emptyMap());
 		final AtomicBoolean done = new AtomicBoolean();
 		willAnswer(i -> {
 			if (done.compareAndSet(false, true)) {
 				return records;
 			}
 			else {
 				Thread.sleep(500);
-				return null;
+				return empty;
 			}
 		}).given(consumer).poll(anyLong());
 		ConsumerFactory cf = mock(ConsumerFactory.class);
@@ -149,15 +159,34 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		}).given(pf).createProducer();
 		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
 		ContainerProperties props = new ContainerProperties("foo");
+		props.setAckMode(ackMode);
 		props.setGroupId("group");
 		props.setTransactionManager(tm);
 		final KafkaTemplate template = new KafkaTemplate(pf);
-		props.setMessageListener((MessageListener) m -> {
-			template.send("bar", "baz");
-			if (handleError) {
-				throw new RuntimeException("fail");
+		if (AckMode.MANUAL_IMMEDIATE.equals(ackMode)) {
+			class AckListener implements AcknowledgingMessageListener {
+				// not a lambda https://bugs.openjdk.java.net/browse/JDK-8074381
+
+				@Override
+				public void onMessage(Object data, Acknowledgment acknowledgment) {
+					template.send("bar", "baz");
+					if (handleError) {
+						throw new RuntimeException("fail");
+					}
+					acknowledgment.acknowledge();
+				}
+
 			}
-		});
+			props.setMessageListener(new AckListener());
+		}
+		else {
+			props.setMessageListener((MessageListener) m -> {
+				template.send("bar", "baz");
+				if (handleError) {
+					throw new RuntimeException("fail");
+				}
+			});
+		}
 		if (handleError) {
 			props.setErrorHandler((e, data) -> {
 			});