GH-689: Support producer-only transactions

Resolves #689

docs/src/main/asciidoc/overview.adoc

@@ -506,6 +506,50 @@ public class Application {
 }
 ----
 
+[[kafka-transactional-binder]]
+=== Transactional Binder
+
+Enable transactions by setting `spring.cloud.stream.kafka.binder.transaction.transactionIdPrefix` to a non-empty value, e.g. `tx-`.
+When used in a processor application, the consumer starts the transaction; any records sent on the consumer thread participate in the same transaction.
+When the listener exits normally, the listener container will send the offset to the transaction and commit it.
+A common producer factory is used for all producer bindings configure using `spring.cloud.stream.kafka.binder.transaction.producer.*` properties; individual binding Kafka producer properties are ignored.
+
+If you wish to use transactions in a source application, or from some arbitrary thread for producer-only transaction (e.g. `@Scheduled` method), you must get a reference to the transactional producer factory and define a `KafkaTransactionManager` bean using it.
+
+====
+[source, java]
+----
+@Bean
+public PlatformTransactionManager transactionManager(BinderFactory binders) {
+    ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null,
+            MessageChannel.class)).getTransactionalProducerFactory();
+    return new KafkaTransactionManager<>(pf);
+}
+----
+====
+
+Notice that we get a reference to the binder using the `BinderFactory`; use `null` in the first argument when there is only one binder configured.
+If more than one binder is configured, use the binder name to get the reference.
+Once we have a reference to the binder, we can obtain a reference to the `ProducerFactory` and create a transaction manager.
+
+Then you would just normal Spring transaction support, e.g. `TransactionTemplate` or `@Transactional`, for example:
+
+====
+[source, java]
+----
+public static class Sender {
+
+    @Transactional
+    public void doInTransaction(MessageChannel output, List<String> stuffToSend) {
+        stuffToSend.forEach(stuff -> output.send(new GenericMessage<>(stuff)));
+    }
+
+}
+----
+====
+
+If you wish to synchronize producer-only transactions with those from some other transaction manager, use a `ChainedTransactionManager`.
+
 [[kafka-error-channels]]
 === Error Channels
 

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -105,6 +105,7 @@ import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
@@ -287,6 +288,17 @@ public class KafkaMessageChannelBinder extends
 		return this.extendedBindingProperties.getExtendedPropertiesEntryClass();
 	}
 
+	/**
+	 * Return a reference to the binder's transaction manager's producer factory (if
+	 * configured). Use this to create a transaction manager in a bean definition when you
+	 * wish to use producer-only transactions.
+	 * @return the transaction manager, or null.
+	 */
+	@Nullable
+	public ProducerFactory<byte[], byte[]> getTransactionalProducerFactory() {
+		return this.transactionManager == null ? null : this.transactionManager.getProducerFactory();
+	}
+
 	@Override
 	protected MessageHandler createProducerMessageHandler(
 			final ProducerDestination destination,

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2019-2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.integration;
+
+import java.util.Map;
+
+import kafka.server.KafkaConfig;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.requests.IsolationLevel;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binder.BinderFactory;
+import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
+import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.kafka.transaction.KafkaTransactionManager;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.AbstractPlatformTransactionManager;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Gary Russell
+ * @since 2.1.4
+ *
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {
+		"spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=tx.",
+		"spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=99",
+		"spring.cloud.stream.kafka.binder.transaction.producer.configuration.acks=all"})
+public class ProducerOnlyTransactionTests {
+
+	private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";
+
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "output")
+			.brokerProperty(KafkaConfig.TransactionsTopicReplicationFactorProp(), "1")
+			.brokerProperty(KafkaConfig.TransactionsTopicMinISRProp(), "1");
+
+	@Autowired
+	private Sender sender;
+
+	@Autowired
+	private MessageChannel output;
+
+	@BeforeClass
+	public static void setup() {
+		System.setProperty(KAFKA_BROKERS_PROPERTY,
+				embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+	}
+
+	@AfterClass
+	public static void clean() {
+		System.clearProperty(KAFKA_BROKERS_PROPERTY);
+	}
+
+	@Test
+	public void testProducerTx() {
+		this.sender.DoInTransaction(this.output);
+		assertThat(this.sender.isInTx()).isTrue();
+		Map<String, Object> props = KafkaTestUtils.consumerProps("consumeTx", "false",
+				embeddedKafka.getEmbeddedKafka());
+		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.name().toLowerCase());
+		Consumer<?, ?> consumer = new KafkaConsumer<>(props);
+		embeddedKafka.getEmbeddedKafka().consumeFromAllEmbeddedTopics(consumer);
+		ConsumerRecord<?, ?> record = KafkaTestUtils.getSingleRecord(consumer, "output");
+		assertThat(record.value()).isEqualTo("foo".getBytes());
+	}
+
+	@EnableBinding(Source.class)
+	@EnableAutoConfiguration
+	@EnableTransactionManagement
+	public static class Config {
+
+		@Bean
+		public PlatformTransactionManager transactionManager(BinderFactory binders) {
+			try {
+				ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null,
+						MessageChannel.class)).getTransactionalProducerFactory();
+				KafkaTransactionManager<byte[], byte[]> tm = new KafkaTransactionManager<>(pf);
+				tm.setTransactionSynchronization(AbstractPlatformTransactionManager.SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
+				return tm;
+			}
+			catch (BeanCreationException e) { // needed to avoid other tests in this package failing when there is no binder
+				return null;
+			}
+		}
+
+		@Bean
+		public Sender sender() {
+			return new Sender();
+		}
+
+	}
+
+	public static class Sender {
+
+		private boolean isInTx;
+
+		@Transactional
+		public void DoInTransaction(MessageChannel output) {
+			this.isInTx = TransactionSynchronizationManager.isActualTransactionActive();
+			output.send(new GenericMessage<>("foo"));
+		}
+
+		public boolean isInTx() {
+			return this.isInTx;
+		}
+
+	}
+
+}