Support transaction initiation in ob endpoints

* Use template.isTransactional()

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java

@@ -62,6 +62,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 import org.springframework.util.concurrent.ListenableFuture;
@@ -94,15 +95,17 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 	private final boolean isGateway;
 
+	private final boolean transactional;
+
 	private EvaluationContext evaluationContext;
 
-	private volatile Expression topicExpression;
+	private Expression topicExpression;
 
-	private volatile Expression messageKeyExpression;
+	private Expression messageKeyExpression;
 
-	private volatile Expression partitionIdExpression;
+	private Expression partitionIdExpression;
 
-	private volatile Expression timestampExpression;
+	private Expression timestampExpression;
 
 	private boolean sync;
 
@@ -141,6 +144,11 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 		else {
 			this.headerMapper = new SimpleKafkaHeaderMapper();
 		}
+		this.transactional = kafkaTemplate.isTransactional();
+		if (this.transactional && this.isGateway) {
+			logger.warn("The KafkaTemplate is transactional; this gateway will only work if the consumer is "
+					+ "configured to read uncommitted records");
+		}
 	}
 
 	public void setTopicExpression(Expression topicExpression) {
@@ -367,7 +375,14 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 			sendFuture = gatewayFuture.getSendFuture();
 		}
 		else {
-			sendFuture = this.kafkaTemplate.send(producerRecord);
+			if (this.transactional && !TransactionSynchronizationManager.isActualTransactionActive()) {
+				sendFuture = this.kafkaTemplate.executeInTransaction(t -> {
+					return t.send(producerRecord);
+				});
+			}
+			else {
+				sendFuture = this.kafkaTemplate.send(producerRecord);
+			}
 			// TODO: In 3.1, always use the success channel.
 			if (!this.noOutputChannel) {
 				metadataChannel = getOutputChannel();

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java

@@ -18,6 +18,10 @@ package org.springframework.integration.kafka.outbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.springframework.kafka.test.assertj.KafkaConditions.key;
 import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
@@ -33,6 +37,8 @@ import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -40,8 +46,10 @@ import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
 import org.junit.Test;
+import org.mockito.InOrder;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.expression.FunctionExpression;
@@ -67,6 +75,7 @@ import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.SettableListenableFuture;
 
@@ -377,4 +386,23 @@ public class KafkaProducerMessageHandlerTests {
 		KafkaTestUtils.getSingleRecord(consumer, topic6);
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testTransaction() {
+		ProducerFactory pf = mock(ProducerFactory.class);
+		given(pf.transactionCapable()).willReturn(true);
+		Producer producer = mock(Producer.class);
+		given(pf.createProducer()).willReturn(producer);
+		ListenableFuture future = mock(ListenableFuture.class);
+		willReturn(future).given(producer).send(any(ProducerRecord.class), any(Callback.class));
+		KafkaTemplate template = new KafkaTemplate(pf);
+		KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template);
+		handler.setTopicExpression(new LiteralExpression("bar"));
+		handler.handleMessage(new GenericMessage<>("foo"));
+		InOrder inOrder = inOrder(producer);
+		inOrder.verify(producer).beginTransaction();
+		inOrder.verify(producer).send(any(ProducerRecord.class), any(Callback.class));
+		inOrder.verify(producer).commitTransaction();
+	}
+
 }