From 72e2aeec2adbc3c161f557c07ce8f07ffa04bfaf Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Wed, 14 Feb 2018 10:17:35 -0500
Subject: [PATCH] GH-305: Fix producer initiated transactions

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`.
---
 .../kafka/KafkaMessageChannelBinder.java      | 27 -------------------
 1 file changed, 27 deletions(-)

diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index 59698bafc..52b74c915 100644
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -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 {