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

@@ -374,6 +374,12 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 			ConsumerRebalanceListener rebalanceListener = createRebalanceListener(consumer);
 
+			if (this.transactionManager != null) {
+				this.transactionTemplate = new TransactionTemplate(this.transactionManager);
+			}
+			else {
+				this.transactionTemplate = null;
+			}
 			if (KafkaMessageListenerContainer.this.topicPartitions == null) {
 				if (this.containerProperties.getTopicPattern() != null) {
 					consumer.subscribe(this.containerProperties.getTopicPattern(), rebalanceListener);
@@ -424,12 +430,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;
@@ -512,18 +512,29 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 						}
 						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(offsets,
-											ListenerConsumer.this.consumerGroupId);
-								}
-
-							});
+							try {
+								offsets.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(offsets);

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

@@ -29,7 +29,9 @@
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -67,6 +69,7 @@
 import org.springframework.kafka.core.ProducerFactoryUtils;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
@@ -132,14 +135,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);
 		ContainerProperties props = new ContainerProperties("foo");
 		props.setGroupId("group");
@@ -161,6 +168,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"));
@@ -169,7 +181,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" })
@@ -409,7 +424,7 @@ public void testRollbackRecord() throws Exception {
 			}
 		});
 
-		@SuppressWarnings({ "rawtypes", "unchecked" })
+		@SuppressWarnings({ "rawtypes" })
 		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
 		containerProps.setTransactionManager(tm);
 		KafkaMessageListenerContainer<Integer, String> container =