Master to 3.3; support non-Tx if template allows

* SI to 5.3.0.B-S

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

@@ -110,6 +110,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 
 	private final boolean transactional;
 
+	private final boolean allowNonTransactional;
+
 	private final AtomicBoolean running = new AtomicBoolean();
 
 	private EvaluationContext evaluationContext;
@@ -164,6 +166,7 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 			this.headerMapper = new SimpleKafkaHeaderMapper();
 		}
 		this.transactional = kafkaTemplate.isTransactional();
+		this.allowNonTransactional = kafkaTemplate.isAllowNonTransactional();
 		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");
@@ -372,7 +375,9 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 	@Override
 	public void stop() {
 		if (this.running.compareAndSet(true, false)) {
-			this.kafkaTemplate.flush();
+			if (!this.transactional || this.allowNonTransactional) {
+				this.kafkaTemplate.flush();
+			}
 		}
 	}
 
@@ -401,9 +406,10 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
 		}
 		else {
 			if (this.transactional
-					&& TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null) {
-				sendFuture = this.kafkaTemplate.executeInTransaction(t -> {
-					return t.send(producerRecord);
+					&& TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null
+					&& !this.allowNonTransactional) {
+				sendFuture = this.kafkaTemplate.executeInTransaction(template -> {
+					return template.send(producerRecord);
 				});
 			}
 			else {

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,7 +18,6 @@ package org.springframework.integration.kafka.config.xml;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
-import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
 
@@ -107,7 +106,7 @@ class KafkaOutboundAdapterParserTests {
 				};
 		@SuppressWarnings("unchecked")
 		ProducerFactory<Integer, String> pf = mock(ProducerFactory.class);
-		given(pf.createProducer(isNull())).willReturn(mockProducer);
+		given(pf.createProducer()).willReturn(mockProducer);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
 		handler.setBeanFactory(mock(BeanFactory.class));

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@ import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.BDDMockito.willReturn;
 import static org.mockito.Mockito.inOrder;
 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;
@@ -474,7 +475,6 @@ class KafkaProducerMessageHandlerTests {
 		inOrder.verify(producer).beginTransaction();
 		inOrder.verify(producer).send(any(ProducerRecord.class), any(Callback.class));
 		inOrder.verify(producer).commitTransaction();
-		inOrder.verify(producer).flush();
 	}
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -584,7 +584,6 @@ class KafkaProducerMessageHandlerTests {
 		inOrder.verify(producer).beginTransaction();
 		inOrder.verify(producer).send(any(ProducerRecord.class), any(Callback.class));
 		inOrder.verify(producer).commitTransaction();
-		inOrder.verify(producer).flush();
 		assertThat(txId.get()).isEqualTo("overridden.tx.id.");
 	}
 
@@ -674,4 +673,19 @@ class KafkaProducerMessageHandlerTests {
 		assertThat(txId.get()).isEqualTo("tm.tx.id.");
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	void testTxNonTx() {
+		KafkaTemplate<String, String> template = mock(KafkaTemplate.class);
+		given(template.isTransactional()).willReturn(true);
+		given(template.inTransaction()).willReturn(false);
+		given(template.isAllowNonTransactional()).willReturn(true);
+		given(template.getProducerFactory()).willReturn(mock(ProducerFactory.class));
+		KafkaProducerMessageHandler<String, String> handler = new KafkaProducerMessageHandler<>(template);
+		handler.setTopicExpression(new LiteralExpression("topic"));
+		handler.handleMessage(new GenericMessage<>("foo"));
+		verify(template, never()).executeInTransaction(any());
+		verify(template).send(any(ProducerRecord.class));
+	}
+
 }