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 {