GH-1388: Support non-Tx with a Tx Template

Resolves #1388

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -193,9 +193,10 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 		String txId = (String) this.configs.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
 		if (StringUtils.hasText(txId)) {
 			setTransactionIdPrefix(txId);
-			LOGGER.info(() -> "If 'setTransactionIdPrefix()' is not going to be configured, "
+			LOGGER.info(() -> "If 'setTransactionIdPrefix()' is not configured, "
 					+ "the existing 'transactional.id' config with value: '" + txId
-					+ "' will be suffixed for concurrent transactions support.");
+					+ "' will be suffixed for concurrent transaction support.");
+			this.configs.remove(ProducerConfig.TRANSACTIONAL_ID_CONFIG);
 		}
 	}
 
@@ -370,6 +371,15 @@ public Producer<K, V> createProducer() {
 	@Override
 	public Producer<K, V> createProducer(@Nullable String txIdPrefixArg) {
 		String txIdPrefix = txIdPrefixArg == null ? this.transactionIdPrefix : txIdPrefixArg;
+		return doCreateProducer(txIdPrefix);
+	}
+
+	@Override
+	public Producer<K, V> createNonTransactionalProducer() {
+		return doCreateProducer(null);
+	}
+
+	private Producer<K, V> doCreateProducer(@Nullable String txIdPrefix) {
 		if (txIdPrefix != null) {
 			if (this.producerPerConsumerPartition) {
 				return createTransactionalProducerForPartition(txIdPrefix);

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-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.
@@ -83,6 +83,8 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 
 	private Duration closeTimeout = ProducerFactoryUtils.DEFAULT_CLOSE_TIMEOUT;
 
+	private boolean allowNonTransactional;
+
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
 	 * @param producerFactory the producer factory.
@@ -181,6 +183,15 @@ public void setCloseTimeout(Duration closeTimeout) {
 		this.closeTimeout = closeTimeout;
 	}
 
+	/**
+	 * Set to true to allow a non-transactional send when the template is transactional.
+	 * @param allowNonTransactional true to allow.
+	 * @since 2.4.3
+	 */
+	public void setAllowNonTransactional(boolean allowNonTransactional) {
+		this.allowNonTransactional = allowNonTransactional;
+	}
+
 	/**
 	 * Return the producer factory used by this template.
 	 * @return the factory.
@@ -390,14 +401,6 @@ protected void closeProducer(Producer<K, V> producer, boolean inTx) {
 	 * RecordMetadata}.
 	 */
 	protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {
-		if (this.transactional) {
-			Assert.state(inTransaction(),
-					"No transaction is in process; "
-						+ "possible solutions: run the template operation within the scope of a "
-						+ "template.executeInTransaction() operation, start a transaction with @Transactional "
-						+ "before invoking the template method, "
-						+ "run in a transaction started by a listener container when consuming a record");
-		}
 		final Producer<K, V> producer = getTheProducer();
 		this.logger.trace(() -> "Sending: " + producerRecord);
 		final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();
@@ -450,7 +453,20 @@ public boolean inTransaction() {
 	}
 
 	private Producer<K, V> getTheProducer() {
-		if (this.transactional) {
+		boolean transactionalProducer = this.transactional;
+		if (transactionalProducer) {
+			boolean inTransaction = inTransaction();
+			Assert.state(this.allowNonTransactional || inTransaction,
+					"No transaction is in process; "
+						+ "possible solutions: run the template operation within the scope of a "
+						+ "template.executeInTransaction() operation, start a transaction with @Transactional "
+						+ "before invoking the template method, "
+						+ "run in a transaction started by a listener container when consuming a record");
+			if (!inTransaction) {
+				transactionalProducer = false;
+			}
+		}
+		if (transactionalProducer) {
 			Producer<K, V> producer = this.producers.get();
 			if (producer != null) {
 				return producer;
@@ -459,8 +475,11 @@ private Producer<K, V> getTheProducer() {
 					.getTransactionalResourceHolder(this.producerFactory, this.transactionIdPrefix, this.closeTimeout);
 			return holder.getProducer();
 		}
+		else if (this.allowNonTransactional) {
+			return this.producerFactory.createNonTransactionalProducer();
+		}
 		else {
-			return this.producerFactory.createProducer(this.transactionIdPrefix);
+			return this.producerFactory.createProducer();
 		}
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.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.
@@ -29,8 +29,9 @@
 public interface ProducerFactory<K, V> {
 
 	/**
-	 * Create a producer.
+	 * Create a producer which will be transactional if the factory is so configured.
 	 * @return the producer.
+	 * @see #transactionCapable()
 	 */
 	Producer<K, V> createProducer();
 
@@ -44,6 +45,16 @@ default Producer<K, V> createProducer(@SuppressWarnings("unused") String txIdPre
 		throw new UnsupportedOperationException("This factory does not support this method");
 	}
 
+	/**
+	 * Create a non-transactional producer.
+	 * @return the producer.
+	 * @since 2.4.3
+	 * @see #transactionCapable()
+	 */
+	default Producer<K, V> createNonTransactionalProducer() {
+		throw new UnsupportedOperationException("This factory does not support this method");
+	}
+
 	/**
 	 * Return true if the factory supports transactions.
 	 * @return true if transactional.

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-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.
@@ -526,6 +526,30 @@ protected Producer<Object, Object> createTransactionalProducerForPartition(Strin
 		}
 	}
 
+	@Test
+	void testNonTxWithTx() {
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(this.embeddedKafka);
+		senderProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx.");
+		senderProps.put(ProducerConfig.RETRIES_CONFIG, 2);
+		DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		pf.setKeySerializer(new StringSerializer());
+		KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true);
+		template.executeInTransaction(tmp -> tmp.execute(prod -> {
+			assertThat(KafkaTestUtils.getPropertyValue(prod, "delegate.transactionManager.transactionalId"))
+					.isEqualTo("tx.0");
+			return null;
+		}));
+		assertThatIllegalStateException().isThrownBy(() -> template.execute(prod -> {
+			return null;
+		}));
+		template.setAllowNonTransactional(true);
+		template.execute(prod -> {
+			assertThat(KafkaTestUtils.getPropertyValue(prod, "delegate.transactionManager.transactionalId")).isNull();
+			return null;
+		});
+		pf.destroy();
+	}
+
 	@Configuration
 	@EnableTransactionManagement
 	public static class DeclarativeConfig {

spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-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.
@@ -20,7 +20,6 @@
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
@@ -396,7 +395,7 @@ public void testAggregateOrphansNotStored() throws Exception {
 		given(container.getContainerProperties()).willReturn(properties);
 		ProducerFactory pf = mock(ProducerFactory.class);
 		Producer producer = mock(Producer.class);
-		given(pf.createProducer(isNull())).willReturn(producer);
+		given(pf.createProducer()).willReturn(producer);
 		AtomicReference<byte[]> correlation = new AtomicReference<>();
 		willAnswer(invocation -> {
 			ProducerRecord rec = invocation.getArgument(0);

src/reference/asciidoc/kafka.adoc

@@ -299,6 +299,8 @@ IMPORTANT: When `producerPerThread` is `true`, user code **must** call `closeThr
 This will physically close the producer and remove it from the `ThreadLocal`.
 Calling `reset()` or `destroy()` will not clean up these producers.
 
+Also see <<tx-template-mixed>>.
+
 When creating a `DefaultKafkaProducerFactory`, key and/or value `Serializer` classes can be picked up from configuration by calling the constructor that only takes in a Map of properties (see example in <<kafka-template>>), or `Serializer` instances may be passed to the `DefaultKafkaProducerFactory` constructor (in which case all `Producer` s share the same instances).
 Alternatively you can provide `Supplier<Serializer>` s (starting with version 2.3) that will be used to obtain separate `Serializer` instances for each `Producer`:
 
@@ -2453,6 +2455,16 @@ This value should be the same for all application instances.
 For transactions started by the template (or the transaction manager for `@Transaction`) you should set the property on the template and transaction manager respectively.
 This property must have a different value on each application instance.
 
+[[tx-template-mixed]]
+===== `KafkaTemplate` Transactional and non-Transactional Publishing
+
+Normally, when a `KafkaTemplate` is transactional (configured with a transaction-capable producer factory), transactions are required.
+The transaction can be started by a `TransactionTemplate`, a `@Transactional` method, calling `executeInTransaction`, or by a listener container, when configured with a `KafkaTransactionManager`.
+Any attempt to use the template outside the scope of a transaction results in the template throwing an `IllegalStateException`.
+Starting with version 2.4.3, you can set the template's `allowNonTransactional` property to `true`.
+In that case, the template will allow the operation to run without a transaction, by calling the `ProducerFactory` 's `createNonTransactionalProducer()` method; the producer will be cached, or thread-bound, as normal for reuse.
+See <<producer-factory>>.
+
 [[transactions-batch]]
 ===== Transactions with Batch Listeners
 

src/reference/asciidoc/whats-new.adoc

@@ -25,6 +25,12 @@ See the IMPORTANT note at the end of <<rebalance-listeners>> for more informatio
 
 The `isAckAfterHandle()` default implementation now returns true by default.
 
+[[x24-template]]
+==== KafkaTemplate
+
+The `KafkaTemplate` now supports non-transactional publishing alongside transactional.
+See <<tx-template-mixed>> for more information.
+
 [[x24-agg]]
 ==== AggregatingReplyingKafkaTemplate