Add ChainedKafkaTransactionManager

Enable transaction synchronization without the need for user code to
send the offset(s) to the transaction.

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -62,7 +62,7 @@ import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.LogIfLevelEnabled;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
-import org.springframework.kafka.transaction.KafkaTransactionManager;
+import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -361,9 +361,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 		private final PlatformTransactionManager transactionManager = this.containerProperties.getTransactionManager();
 
 		@SuppressWarnings("rawtypes")
-		private final KafkaTransactionManager kafkaTxManager =
-				this.transactionManager instanceof KafkaTransactionManager
-						? ((KafkaTransactionManager) this.transactionManager) : null;
+		private final KafkaAwareTransactionManager kafkaTxManager =
+				this.transactionManager instanceof KafkaAwareTransactionManager
+						? ((KafkaAwareTransactionManager) this.transactionManager) : null;
 
 		private final TransactionTemplate transactionTemplate;
 

spring-kafka/src/main/java/org/springframework/kafka/transaction/ChainedKafkaTransactionManager.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.kafka.transaction;
+
+import org.springframework.data.transaction.ChainedTransactionManager;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ChainedTransactionManager} that has exactly one
+ * {@link KafkaAwareTransactionManager} in the chain.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Gary Russell
+ * @since 2.1.3
+ *
+ */
+public class ChainedKafkaTransactionManager<K, V> extends ChainedTransactionManager implements KafkaAwareTransactionManager<K, V> {
+
+	private final KafkaAwareTransactionManager<K, V> kafkaTransactionManager;
+
+	@SuppressWarnings("unchecked")
+	public ChainedKafkaTransactionManager(PlatformTransactionManager... transactionManagers) {
+		super(transactionManagers);
+		KafkaAwareTransactionManager<K, V> kafkaTransactionManager = null;
+		for (PlatformTransactionManager tm : transactionManagers) {
+			if (tm instanceof KafkaAwareTransactionManager) {
+				Assert.isNull(kafkaTransactionManager, "Only one KafkaAwareTransactionManager is allowed");
+				kafkaTransactionManager = (KafkaTransactionManager<K, V>) tm;
+			}
+		}
+		Assert.notNull(kafkaTransactionManager, "Exactly one KafkaAwareTransactionManager is required");
+		this.kafkaTransactionManager = kafkaTransactionManager;
+	}
+
+	@Override
+	public ProducerFactory<K, V> getProducerFactory() {
+		return this.kafkaTransactionManager.getProducerFactory();
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaAwareTransactionManager.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.kafka.transaction;
+
+import org.springframework.kafka.core.ProducerFactory;
+
+/**
+ * A transaction manager that can provide a {@link ProducerFactory}.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
+ *
+ * @author Gary Russell
+ * @since 2.1.3
+ *
+ */
+public interface KafkaAwareTransactionManager<K, V> {
+
+	/**
+	 * Get the producer factory.
+	 * @return the producerFactory
+	 */
+	ProducerFactory<K, V> getProducerFactory();
+
+}

spring-kafka/src/main/java/org/springframework/kafka/transaction/KafkaTransactionManager.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -66,7 +66,7 @@ import org.springframework.util.Assert;
  */
 @SuppressWarnings("serial")
 public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionManager
-		implements ResourceTransactionManager {
+		implements ResourceTransactionManager, KafkaAwareTransactionManager<K, V> {
 
 	private final ProducerFactory<K, V> producerFactory;
 
@@ -88,6 +88,7 @@ public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionMa
 	 * Get the producer factory.
 	 * @return the producerFactory
 	 */
+	@Override
 	public ProducerFactory<K, V> getProducerFactory() {
 		return this.producerFactory;
 	}

spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -64,7 +64,9 @@ import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.config.ContainerProperties;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.kafka.transaction.ChainedKafkaTransactionManager;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
+import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionException;
 import org.springframework.transaction.support.AbstractPlatformTransactionManager;
@@ -88,9 +90,18 @@ public class TransactionalContainerTests {
 	@ClassRule
 	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(3, true, topic1, topic2);
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testConsumeAndProduceTransaction() throws Exception {
+	public void testConsumeAndProduceTransactionKTM() throws Exception {
+		testConsumeAndProduceTransactionGuts(false);
+	}
+
+	@Test
+	public void testConsumeAndProduceTransactionKCTM() throws Exception {
+		testConsumeAndProduceTransactionGuts(true);
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	private void testConsumeAndProduceTransactionGuts(boolean chained) throws Exception {
 		Consumer consumer = mock(Consumer.class);
 		final TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
@@ -122,9 +133,13 @@ public class TransactionalContainerTests {
 		given(pf.transactionCapable()).willReturn(true);
 		given(pf.createProducer()).willReturn(producer);
 		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
+		PlatformTransactionManager ptm = tm;
+		if (chained) {
+			ptm = new ChainedKafkaTransactionManager(new SomeOtherTransactionManager(), tm);
+		}
 		ContainerProperties props = new ContainerProperties("foo");
 		props.setGroupId("group");
-		props.setTransactionManager(tm);
+		props.setTransactionManager(ptm);
 		final KafkaTemplate template = new KafkaTemplate(pf);
 		props.setMessageListener((MessageListener) m -> {
 			template.send("bar", "baz");

src/reference/asciidoc/kafka.adoc

@@ -280,7 +280,7 @@ If the listener throws an exception, the transaction is rolled back and the cons
 If you need to synchronize a Kafka transaction with some other transaction; simply configure the listener container with the appropriate transaction manager (one that supports synchronization, such as the `DataSourceTransactionManager`).
 Any operations performed on a **transactional** `KafkaTemplate` from the listener will participate in a single transaction.
 The Kafka transaction will be committed (or rolled back) immediately after the controlling transaction.
-Before exiting the listener, you should invoke one of the template's `sendOffsetsToTransaction` methods.
+Before exiting the listener, you should invoke one of the template's `sendOffsetsToTransaction` methods (unless you use a <<chained-transaction-manager, `ChainedKafkaTransactionManager` - see below>>).
 For convenience, the listener container binds its consumer group id to the thread so, generally, you can use the first method:
 
 [source, java]
@@ -317,6 +317,15 @@ NOTE: The offset to be committed is one greater than the offset of the record(s)
 IMPORTANT: This should only be called when using transaction synchronization.
 When a listener container is configured to use a `KafkaTransactionManager`, it will take care of sending the offsets to the transaction.
 
+[[chained-transaction-manager]]
+====== ChainedKafkaTransactionManager
+
+The `ChainedKafkaTransactionManager` was introduced in _version 2.1.3_.
+This is a subclass of `ChainedTransactionManager` that can have exactly one `KafkaTransactionManager`.
+Since it is a `KafkaAwareTransactionManager`, the container can send the offsets to the transaction in the same way as when the container is configured with a simple `KafkaTransactionManager`.
+This provides another mechanism for synchronizing transactions without having to send the offsets to the transaction in the listener code.
+Chain your transaction managers in the desired order and provide the `ChainedTransactionManager` in the `ContainerProperties`.
+
 ====== KafkaTemplate Local Transactions
 
 You can use the `KafkaTemplate` to execute a series of operations within a local transaction.

src/reference/asciidoc/whats-new.adoc

@@ -42,6 +42,10 @@ See <<class-level-kafkalistener>> for more information.
 Starting with _version 2.1.3_, a subclass of `KafkaTemplate` is provided to support request/reply semantics.
 See <<replying-template>> for more information.
 
+==== ChainedKafkaTransactionManager
+
+_version 2.1.3_ introduced the `ChainedKafkaTransactionManager` see <<chained-transaction-manager>> for more information.
+
 ==== Migration Guide from 2.0
 
 https://github.com/spring-projects/spring-kafka/wiki/Spring-for-Apache-Kafka-2.0-to-2.1-Migration-Guide[2.0 to 2.1 Migration].