SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-305: Fix producer initiated transactions

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/305
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/308

Requires https://github.com/spring-projects/spring-integration-kafka/pull/196

`ProducerConfigurationMessageHandler` overrode `handleMessageInternal` to support
producer-initiated transactions. Now that the superclass extends `ARPMH`, this
method is no longer overridable.

Spring Integration Kafka's `KafkaProducerMessageHandler` now detects a transactional
`KafkaTemplate` and will automatically start a transaction if needed, so there is no
longer any need to override that method.

No additional tests needed; see `KafkaTransactionTests`.
This commit is contained in:
Gary Russell
2018-02-14 10:17:35 -05:00
committed by Artem Bilan
parent 090125fa71
commit 72e2aeec2a
1 changed files with 0 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -74,7 +74,6 @@ import org.springframework.integration.support.AcknowledgmentCallback;
import org.springframework.integration.support.AcknowledgmentCallback.Status;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.kafka.KafkaException;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -91,14 +90,11 @@ import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.transaction.KafkaTransactionManager;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
@@ -136,8 +132,6 @@ public class KafkaMessageChannelBinder extends
private final KafkaTransactionManager<byte[], byte[]> transactionManager;
private final TransactionTemplate transactionTemplate;
private ProducerListener<byte[], byte[]> producerListener;
private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties();
@@ -150,11 +144,9 @@ public class KafkaMessageChannelBinder extends
this.transactionManager = new KafkaTransactionManager<>(
getProducerFactory(configurationProperties.getTransaction().getTransactionIdPrefix(),
new ExtendedProducerProperties<>(configurationProperties.getTransaction().getProducer())));
this.transactionTemplate = new TransactionTemplate(this.transactionManager);
}
else {
this.transactionManager = null;
this.transactionTemplate = null;
}
}
@@ -733,25 +725,6 @@ public class KafkaMessageChannelBinder extends
return this.running;
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
if (KafkaMessageChannelBinder.this.transactionTemplate != null
&& !TransactionSynchronizationManager.isActualTransactionActive()) {
KafkaMessageChannelBinder.this.transactionTemplate.execute(s -> {
try {
super.handleMessageInternal(message);
}
catch (Exception e) {
throw new KafkaException("Exception on transactional send", e);
}
return null;
});
}
else {
super.handleMessageInternal(message);
}
}
}
static class TopicInformation {