Refactor KafkaAwareTransactionManager

Sub-interface of `PlatformTransactionManager` to facilitate Boot auto-configuration.

Deprecate `ResourceTransactionManager` inheritance.

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

@@ -55,4 +55,18 @@ public class ChainedKafkaTransactionManager<K, V> extends ChainedTransactionMana
 		return this.kafkaTransactionManager.getProducerFactory();
 	}
 
+	/**
+	 * Return the producer factory.
+	 * @return the producer factory.
+	 * @deprecated - in a future release {@link KafkaAwareTransactionManager} will not be
+	 * a sub interface of
+	 * {@link org.springframework.transaction.support.ResourceTransactionManager}.
+	 * TODO: Remove in 3.0
+	 */
+	@Deprecated
+	@Override
+	public Object getResourceFactory() {
+		return getProducerFactory();
+	}
+
 }

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

@@ -17,9 +17,12 @@
 package org.springframework.kafka.transaction;
 
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.transaction.support.ResourceTransactionManager;
 
 /**
  * A transaction manager that can provide a {@link ProducerFactory}.
+ * Currently a sub-interface of {@link ResourceTransactionManager}
+ * for backwards compatibility.
  *
  * @param <K> the key type.
  * @param <V> the value type.
@@ -28,7 +31,7 @@ import org.springframework.kafka.core.ProducerFactory;
  * @since 2.1.3
  *
  */
-public interface KafkaAwareTransactionManager<K, V> {
+public interface KafkaAwareTransactionManager<K, V> extends ResourceTransactionManager {
 
 	/**
 	 * Get the producer factory.

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

@@ -66,7 +66,7 @@ import org.springframework.util.Assert;
  */
 @SuppressWarnings("serial")
 public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionManager
-		implements ResourceTransactionManager, KafkaAwareTransactionManager<K, V> {
+		implements KafkaAwareTransactionManager<K, V> {
 
 	private final ProducerFactory<K, V> producerFactory;
 
@@ -93,6 +93,14 @@ public class KafkaTransactionManager<K, V> extends AbstractPlatformTransactionMa
 		return this.producerFactory;
 	}
 
+	/**
+	 * Return the producer factory.
+	 * @return the producer factory.
+	 * @deprecated - in a future release {@link KafkaAwareTransactionManager} will
+	 * not be a sub interface of {@link ResourceTransactionManager}.
+	 * TODO: Remove in 3.0
+	 */
+	@Deprecated
 	@Override
 	public Object getResourceFactory() {
 		return getProducerFactory();

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -47,9 +47,11 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mockito;
 
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
 import org.springframework.context.event.EventListener;
 import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
 import org.springframework.core.convert.converter.Converter;
@@ -88,6 +90,9 @@ import org.springframework.kafka.support.converter.StringJsonMessageConverter;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.kafka.transaction.ChainedKafkaTransactionManager;
+import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
+import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.lang.NonNull;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -173,6 +178,9 @@ public class EnableKafkaIntegrationTests {
 	@Autowired
 	private FooConverter fooConverter;
 
+	@Autowired
+	private ConcurrentKafkaListenerContainerFactory<Integer, String> transactionalFactory;
+
 	@Test
 	public void testAnonymous() {
 		MessageListenerContainer container = this.registry
@@ -669,6 +677,12 @@ public class EnableKafkaIntegrationTests {
 		assertThat(this.listener.consumerRecords.iterator().next().value()).isEqualTo("allRecords");
 	}
 
+	@Test
+	public void testAutoConfigTm() {
+		assertThat(this.transactionalFactory.getContainerProperties().getTransactionManager())
+			.isInstanceOf(ChainedKafkaTransactionManager.class);
+	}
+
 	@Configuration
 	@EnableKafka
 	@EnableTransactionManagement(proxyTargetClass = true)
@@ -686,11 +700,23 @@ public class EnableKafkaIntegrationTests {
 			return Mockito.mock(PlatformTransactionManager.class);
 		}
 
+		@Bean
+		public KafkaTransactionManager<Integer, String> ktm() {
+			return new KafkaTransactionManager<>(txProducerFactory());
+		}
+
+		@Bean
+		@Primary
+		public ChainedKafkaTransactionManager<Integer, String> cktm() {
+			return new ChainedKafkaTransactionManager<>(ktm(), transactionManager());
+		}
+
 		private Throwable globalErrorThrowable;
 
 		@Bean
 		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
-		kafkaListenerContainerFactory() {
+				kafkaListenerContainerFactory() {
+
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory());
@@ -705,13 +731,28 @@ public class EnableKafkaIntegrationTests {
 
 		@Bean
 		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
-		withNoReplyTemplateContainerFactory() {
+				withNoReplyTemplateContainerFactory() {
+
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory());
 			return factory;
 		}
 
+		@Bean
+		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
+				transactionalFactory(ObjectProvider<KafkaAwareTransactionManager<Integer, String>> tm) {
+
+			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(consumerFactory());
+			KafkaAwareTransactionManager<Integer, String> ktm = tm.getIfUnique();
+			if (ktm != null) {
+				factory.getContainerProperties().setTransactionManager(ktm);
+			}
+			return factory;
+		}
+
 		@Bean
 		public RecordPassAllFilter recordFilter() {
 			return new RecordPassAllFilter();
@@ -926,6 +967,13 @@ public class EnableKafkaIntegrationTests {
 			return new DefaultKafkaProducerFactory<>(producerConfigs());
 		}
 
+		@Bean
+		public ProducerFactory<Integer, String> txProducerFactory() {
+			DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(producerConfigs());
+			pf.setTransactionIdPrefix("tx-");
+			return pf;
+		}
+
 		@Bean
 		public Map<String, Object> producerConfigs() {
 			return KafkaTestUtils.producerProps(embeddedKafka);
@@ -1481,7 +1529,7 @@ public class EnableKafkaIntegrationTests {
 		}
 
 		@KafkaListener(id = "ifctx", topics = "annotated9")
-		@Transactional
+		@Transactional(transactionManager = "transactionManager")
 		public void listenTx(String foo) {
 			latch2.countDown();
 		}