Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-900: Use correct txId for initial commit

Browse Source 
Fixes https://github.com/spring-projects/spring-kafka/issues/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.
This commit is contained in:
Gary Russell
2018-12-10 16:20:24 -05:00
parent 7891bc04bb
commit 87ec5829fd
2 changed files with 46 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -374,6 +374,12 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
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 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
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 class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
}
if (ListenerConsumer.this.transactionTemplate != null &&
ListenerConsumer.this.kafkaTxManager != null) {
ListenerConsumer.this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
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(offsets,
ListenerConsumer.this.consumerGroupId);
}
@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);

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

@@ -29,7 +29,9 @@ import static org.mockito.Mockito.mock;
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.ProducerFactory;
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 @@ public class TransactionalContainerTests {
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 @@ public class TransactionalContainerTests {
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 @@ public class TransactionalContainerTests {
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 class TransactionalContainerTests {
}
});
@SuppressWarnings({ "rawtypes", "unchecked" })
@SuppressWarnings({ "rawtypes" })
KafkaTransactionManager tm = new KafkaTransactionManager(pf);
containerProps.setTransactionManager(tm);
KafkaMessageListenerContainer<Integer, String> container =