GH-900: Use correct txId for initial commit

Fixes #900

Use the proper zombie-fencing `transational.id` when committing initial
offsets after assignment.

**cherry-pick to all, 1.3.x will probable need a backport**

* Fix - set TL before invoking the transaction template.

Author: garyrussell

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

@@ -490,6 +490,12 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 							this.containerProperties.getClientId(),
 							KafkaMessageListenerContainer.this.clientIdSuffix);
 
+			if (this.transactionManager != null) {
+				this.transactionTemplate = new TransactionTemplate(this.transactionManager);
+			}
+			else {
+				this.transactionTemplate = null;
+			}
 			subscribeOrAssignTopics(this.consumer);
 			GenericErrorHandler<?> errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler();
 			this.genericListener = listener;
@@ -525,12 +531,6 @@ else if (listener instanceof MessageListener) {
 				this.batchErrorHandler = new BatchLoggingErrorHandler();
 			}
 			Assert.state(!this.isBatchListener || !this.isRecordAck, "Cannot use AckMode.RECORD with a batch listener");
-			if (this.transactionManager != null) {
-				this.transactionTemplate = new TransactionTemplate(this.transactionManager);
-			}
-			else {
-				this.transactionTemplate = null;
-			}
 			if (this.containerProperties.getScheduler() != null) {
 				this.taskScheduler = this.containerProperties.getScheduler();
 				this.taskSchedulerExplicitlySet = true;
@@ -1654,18 +1654,29 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 					ListenerConsumer.this.commitLogger.log(() -> "Committing on assignment: " + offsetsToCommit);
 					if (ListenerConsumer.this.transactionTemplate != null &&
 							ListenerConsumer.this.kafkaTxManager != null) {
-						ListenerConsumer.this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
-
-							@SuppressWarnings({ "unchecked", RAWTYPES })
-							@Override
-							protected void doInTransactionWithoutResult(TransactionStatus status) {
-								((KafkaResourceHolder) TransactionSynchronizationManager
-										.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
-										.getProducer().sendOffsetsToTransaction(offsetsToCommit, // NOSONAR
-												ListenerConsumer.this.consumerGroupId);
-							}
-
-						});
+						try {
+							offsetsToCommit.forEach((partition, offsetAndMetadata) -> {
+								TransactionSupport.setTransactionIdSuffix(
+										zombieFenceTxIdSuffix(partition.topic(), partition.partition()));
+								ListenerConsumer.this.transactionTemplate
+										.execute(new TransactionCallbackWithoutResult() {
+
+									@SuppressWarnings({ "unchecked", RAWTYPES })
+									@Override
+									protected void doInTransactionWithoutResult(TransactionStatus status) {
+										((KafkaResourceHolder) TransactionSynchronizationManager
+												.getResource(ListenerConsumer.this.kafkaTxManager.getProducerFactory()))
+												.getProducer().sendOffsetsToTransaction(
+														Collections.singletonMap(partition, offsetAndMetadata),
+														ListenerConsumer.this.consumerGroupId);
+									}
+
+								});
+							});
+						}
+						finally {
+							TransactionSupport.clearTransactionIdSuffix();
+						}
 					}
 					else if (KafkaMessageListenerContainer.this.getContainerProperties().isSyncCommits()) {
 						ListenerConsumer.this.consumer.commitSync(offsetsToCommit);

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

@@ -30,8 +30,10 @@
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -73,6 +75,7 @@
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -153,14 +156,18 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		ConsumerFactory cf = mock(ConsumerFactory.class);
 		willReturn(consumer).given(cf).createConsumer("group", "", null);
 		Producer producer = mock(Producer.class);
-		final CountDownLatch closeLatch = new CountDownLatch(1);
+		final CountDownLatch closeLatch = new CountDownLatch(2);
 		willAnswer(i -> {
 			closeLatch.countDown();
 			return null;
 		}).given(producer).close();
 		ProducerFactory pf = mock(ProducerFactory.class);
 		given(pf.transactionCapable()).willReturn(true);
-		given(pf.createProducer()).willReturn(producer);
+		final List<String> transactionalIds = new ArrayList<>();
+		willAnswer(i -> {
+			transactionalIds.add(TransactionSupport.getTransactionIdSuffix());
+			return producer;
+		}).given(pf).createProducer();
 		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
 		PlatformTransactionManager ptm = tm;
 		if (chained) {
@@ -185,6 +192,11 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		assertThat(closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		InOrder inOrder = inOrder(producer);
 		inOrder.verify(producer).beginTransaction();
+		inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
+				new OffsetAndMetadata(0)), "group");
+		inOrder.verify(producer).commitTransaction();
+		inOrder.verify(producer).close();
+		inOrder.verify(producer).beginTransaction();
 		ArgumentCaptor<ProducerRecord> captor = ArgumentCaptor.forClass(ProducerRecord.class);
 		inOrder.verify(producer).send(captor.capture(), any(Callback.class));
 		assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
@@ -193,7 +205,10 @@ private void testConsumeAndProduceTransactionGuts(boolean chained, boolean handl
 		inOrder.verify(producer).commitTransaction();
 		inOrder.verify(producer).close();
 		container.stop();
-		verify(pf, times(1)).createProducer();
+		verify(pf, times(2)).createProducer();
+		verifyNoMoreInteractions(producer);
+		assertThat(transactionalIds.get(0)).isEqualTo("group.foo.0");
+		assertThat(transactionalIds.get(0)).isEqualTo("group.foo.0");
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
@@ -433,7 +448,7 @@ public void testRollbackRecord() throws Exception {
 			}
 		});
 
-		@SuppressWarnings({ "rawtypes", "unchecked" })
+		@SuppressWarnings({ "rawtypes" })
 		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
 		containerProps.setTransactionManager(tm);
 		KafkaMessageListenerContainer<Integer, String> container =