GH-2775: Deprecate CP#transactionManager property

Fixes: #2775

* Use the more specific `KafkaAwareTransactionManager` type instead of the `PlatfromTransactionManager`
  to properly manage Kafka transactions.
* Deprecate transactionManager property in `ContainerProperties` and add 'kafkaAwareTransationManager`
  as a new property.
* Modify transaction unit tests to `setKafkaAwareTransactionManager`.

Author: Wzy19930507

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/container-props.adoc

@@ -123,6 +123,10 @@ Also see `idleBeforeDataMultiplier`.
 |None
 |Used to override any arbitrary consumer properties configured on the consumer factory.
 
+|[[kafkaAwareTransactionManager]]<<kafkaAwareTransactionManager,`kafkaAwareTransactionManager`>>
+|`null`
+|See xref:kafka/transactions.adoc[Transactions].
+
 |[[listenerTaskExecutor]]<<listenerTaskExecutor,`listenerTaskExecutor`>>
 |`SimpleAsyncTaskExecutor`
 |A task executor to run the consumer threads.
@@ -232,7 +236,7 @@ Mutually exclusive; at least one must be provided; enforced by `ContainerPropert
 
 |[[transactionManager]]<<transactionManager,`transactionManager`>>
 |`null`
-|See xref:kafka/transactions.adoc[Transactions].
+|Deprecated since 3.2, see <<kafkaAwareTransactionManager>>, xref:kafka/transactions.adoc#transaction-synchronization[Other transaction managers].
 |===
 
 [[alc-props]]

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -36,6 +36,11 @@ Rules for the redirection are set either via the `RetryableTopic.exceptionBasedD
 Custom DLTs are created automatically as well as other retry and dead-letter topics.
 See xref:retrytopic/features.adoc#exc-based-custom-dlt-routing[Routing of messages to custom DLTs based on thrown exceptions] for more information.
 
+[[x32-cp-ptm]]
+=== Deprecating ContainerProperties transactionManager property
+
+Deprecating the `transactionManager` property in `ContainerProperties` in favor of `KafkaAwareTransactionManager`,  a narrower type compared to the general `PlatformTransactionManager`. See xref:kafka/container-props.adoc#kafkaAwareTransactionManager[ContainerProperties] and xref:kafka/transactions.adoc#transaction-synchronization[Transaction Synchronization].
+
 [[x32-after-rollback-processing]]
 === After Rollback Processing
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2023 the original author or authors.
+ * Copyright 2016-2024 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.
@@ -35,6 +35,7 @@
 import org.springframework.core.task.AsyncTaskExecutor;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.micrometer.KafkaListenerObservationConvention;
+import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.lang.Nullable;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.transaction.PlatformTransactionManager;
@@ -257,8 +258,11 @@ public enum EOSMode {
 
 	private double idleBeforeDataMultiplier = DEFAULT_IDLE_BEFORE_DATA_MULTIPLIER;
 
+	@Deprecated(since = "3.2")
 	private PlatformTransactionManager transactionManager;
 
+	private KafkaAwareTransactionManager<?, ?> kafkaAwareTransactionManager;
+
 	private boolean batchRecoverAfterRollback = false;
 
 	private int monitorInterval = DEFAULT_MONITOR_INTERVAL;
@@ -371,7 +375,7 @@ public void setMessageListener(Object messageListener) {
 	 * calling thread and sometimes not.</li>
 	 * </ul>
 	 * @param ackMode the {@link AckMode}; default BATCH.
-	 * @see #setTransactionManager(PlatformTransactionManager)
+	 * @see #setKafkaAwareTransactionManager(KafkaAwareTransactionManager)
 	 */
 	public void setAckMode(AckMode ackMode) {
 		Assert.notNull(ackMode, "'ackMode' cannot be null");
@@ -525,6 +529,7 @@ public Long getIdlePartitionEventInterval() {
 		return this.idlePartitionEventInterval;
 	}
 
+	@Deprecated(since = "3.2", forRemoval = true)
 	@Nullable
 	public PlatformTransactionManager getTransactionManager() {
 		return this.transactionManager;
@@ -542,10 +547,25 @@ public PlatformTransactionManager getTransactionManager() {
 	 * @since 1.3
 	 * @see #setAckMode(AckMode)
 	 */
+	@Deprecated(since = "3.2", forRemoval = true)
 	public void setTransactionManager(@Nullable PlatformTransactionManager transactionManager) {
 		this.transactionManager = transactionManager;
 	}
 
+	@Nullable
+	public KafkaAwareTransactionManager<?, ?> getKafkaAwareTransactionManager() {
+		return this.kafkaAwareTransactionManager;
+	}
+
+	/**
+	 * Set the transaction manager to start a transaction; replace {@link #setTransactionManager}.
+	 * @param kafkaAwareTransactionManager the transaction manager.
+	 * @since 3.2
+	 */
+	public void setKafkaAwareTransactionManager(@Nullable KafkaAwareTransactionManager<?, ?> kafkaAwareTransactionManager) {
+		this.kafkaAwareTransactionManager = kafkaAwareTransactionManager;
+	}
+
 	/**
 	 * Recover batch records after rollback if true.
 	 * @return true to recover.
@@ -857,16 +877,16 @@ public TransactionDefinition getTransactionDefinition() {
 	/**
 	 * Set a transaction definition with properties (e.g. timeout) that will be copied to
 	 * the container's transaction template. Note that this is only generally useful when
-	 * used with a {@link #setTransactionManager(PlatformTransactionManager)
-	 * PlatformTransactionManager} that supports a custom definition; this does NOT
+	 * used with a {@link #setKafkaAwareTransactionManager(KafkaAwareTransactionManager)
+	 * KafkaAwareTransactionManager} that supports a custom definition; this does NOT
 	 * include the {@link org.springframework.kafka.transaction.KafkaTransactionManager}
 	 * which has no concept of transaction timeout. It can be useful to start, for example
 	 * a database transaction, in the container, rather than using {@code @Transactional}
 	 * on the listener, because then a record interceptor, or filter in a listener adapter
 	 * can participate in the transaction.
 	 * @param transactionDefinition the definition.
 	 * @since 2.5.4
-	 * @see #setTransactionManager(PlatformTransactionManager)
+	 * @see #setKafkaAwareTransactionManager(KafkaAwareTransactionManager)
 	 */
 	public void setTransactionDefinition(@Nullable TransactionDefinition transactionDefinition) {
 		this.transactionDefinition = transactionDefinition;
@@ -1072,6 +1092,9 @@ public String toString() {
 				+ (this.transactionManager != null
 						? "\n transactionManager=" + this.transactionManager
 						: "")
+				+ (this.kafkaAwareTransactionManager != null
+						? "\n kafkaAwareTransactionManager=" + this.kafkaAwareTransactionManager
+						: "")
 				+ "\n monitorInterval=" + this.monitorInterval
 				+ (this.scheduler != null ? "\n scheduler=" + this.scheduler : "")
 				+ "\n noPollThreshold=" + this.noPollThreshold

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

@@ -696,12 +696,16 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final CommonErrorHandler commonErrorHandler;
 
-		private final PlatformTransactionManager transactionManager = this.containerProperties.getTransactionManager();
+		@Deprecated(since = "3.2", forRemoval = true)
+		@SuppressWarnings("removal")
+		private final PlatformTransactionManager transactionManager =
+				this.containerProperties.getKafkaAwareTransactionManager() != null ?
+						this.containerProperties.getKafkaAwareTransactionManager() :
+						this.containerProperties.getTransactionManager();
 
-		@SuppressWarnings(RAWTYPES)
-		private final KafkaAwareTransactionManager kafkaTxManager =
-				this.transactionManager instanceof KafkaAwareTransactionManager
-						? ((KafkaAwareTransactionManager) this.transactionManager) : null;
+		private final KafkaAwareTransactionManager<?, ?> kafkaTxManager =
+				this.transactionManager instanceof KafkaAwareTransactionManager<?, ?> kafkaAwareTransactionManager ?
+						kafkaAwareTransactionManager : null;
 
 		private final TransactionTemplate transactionTemplate;
 
@@ -3034,7 +3038,6 @@ private void sendOffsetsToTransaction() {
 			doSendOffsets(this.producer, commits);
 		}
 
-		@SuppressWarnings("deprecation")
 		private void doSendOffsets(Producer<?, ?> prod, Map<TopicPartition, OffsetAndMetadata> commits) {
 			prod.sendOffsetsToTransaction(commits, this.consumer.groupMetadata());
 			if (this.fixTxOffsets) {

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

@@ -355,7 +355,7 @@ public void testNestedTxProducerIsCached() throws Exception {
 			latch.countDown();
 		});
 		KafkaTransactionManager<Integer, String> tm = new KafkaTransactionManager<>(pfTx);
-		containerProps.setTransactionManager(tm);
+		containerProps.setKafkaAwareTransactionManager(tm);
 		KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
 				containerProps);
 		container.start();
@@ -406,7 +406,7 @@ public void testNestedTxProducerIsFixed() throws Exception {
 			latch.countDown();
 		});
 		KafkaTransactionManager<Integer, String> tm = new KafkaTransactionManager<>(pfTx);
-		containerProps.setTransactionManager(tm);
+		containerProps.setKafkaAwareTransactionManager(tm);
 		KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
 				containerProps);
 		container.start();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2024 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.
@@ -120,7 +120,7 @@ void testLatestOnlyTx() throws InterruptedException {
 			latch.countDown();
 			return null;
 		}).given(producer).sendOffsetsToTransaction(any(), any(ConsumerGroupMetadata.class));
-		props.setTransactionManager(tm);
+		props.setKafkaAwareTransactionManager(tm);
 		this.registry.start();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
@@ -135,8 +135,7 @@ void testLatestOnlyNoTx() throws InterruptedException {
 		KafkaTransactionManager tm = new KafkaTransactionManager<>(pf);
 		Producer producer = mock(Producer.class);
 		given(pf.createProducer(any())).willReturn(producer);
-		CountDownLatch latch = new CountDownLatch(1);
-		props.setTransactionManager(tm);
+		props.setKafkaAwareTransactionManager(tm);
 		this.registry.start();
 		assertThat(this.config.commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		verify(producer, never()).sendOffsetsToTransaction(any(), any(ConsumerGroupMetadata.class));

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

@@ -561,7 +561,7 @@ void testIntercept(boolean beforeTx, @Nullable AssignmentCommitOption option, bo
 		given(tm.getProducerFactory()).willReturn(pf);
 		Producer producer = mock(Producer.class);
 		given(pf.createProducer()).willReturn(producer);
-		containerProperties.setTransactionManager(tm);
+		containerProperties.setKafkaAwareTransactionManager(tm);
 		List<String> order = new ArrayList<>();
 		CountDownLatch latch = new CountDownLatch(option == null ? 2 : 3);
 		willAnswer(inv -> {
@@ -688,7 +688,6 @@ void testInterceptInTxNonKafkaTM() throws InterruptedException {
 			.willReturn(consumer);
 		ContainerProperties containerProperties = new ContainerProperties("foo");
 		containerProperties.setGroupId("grp");
-		AtomicReference<List<ConsumerRecord<String, String>>> received = new AtomicReference<>();
 		containerProperties.setMessageListener((MessageListener) rec -> {
 		});
 		containerProperties.setMissingTopicsFatal(false);

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

@@ -3951,7 +3951,7 @@ public ConsumerRecord<Integer, String> intercept(ConsumerRecord<Integer, String>
 
 	@ParameterizedTest(name = "{index} testInvokeBatchInterceptorAllSkipped early intercept {0}")
 	@ValueSource(booleans = { true, false })
-	@SuppressWarnings({"unchecked"})
+	@SuppressWarnings({ "unchecked", "deprecation" })
 	public void testInvokeBatchInterceptorAllSkipped(boolean early) throws Exception {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
 		Consumer<Integer, String> consumer = mock(Consumer.class);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -279,7 +279,7 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			factory.setConsumerFactory(consumerFactory());
 			factory.getContainerProperties().setAckMode(AckMode.MANUAL);
 			factory.getContainerProperties().setMissingTopicsFatal(false);
-			factory.getContainerProperties().setTransactionManager(tm());
+			factory.getContainerProperties().setKafkaAwareTransactionManager(tm());
 			factory.setBatchListener(true);
 			return factory;
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -247,7 +247,7 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			factory.setConsumerFactory(consumerFactory());
 			factory.setCommonErrorHandler(new DefaultErrorHandler());
 			factory.getContainerProperties().setAckMode(AckMode.BATCH);
-			factory.getContainerProperties().setTransactionManager(tm());
+			factory.getContainerProperties().setKafkaAwareTransactionManager(tm());
 			return factory;
 		}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -248,7 +248,7 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			factory.setConsumerFactory(consumerFactory());
 			factory.setCommonErrorHandler(new DefaultErrorHandler());
 			factory.getContainerProperties().setAckMode(AckMode.RECORD);
-			factory.getContainerProperties().setTransactionManager(tm());
+			factory.getContainerProperties().setKafkaAwareTransactionManager(tm());
 			return factory;
 		}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -62,10 +62,10 @@
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.annotation.DirtiesContext.ClassMode;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
-import org.springframework.transaction.PlatformTransactionManager;
 
 /**
  * @author Gary Russell
@@ -156,7 +156,7 @@ void defaults() {
 
 		containerProps = new ContainerProperties("sbpp");
 		containerProps.setMessageListener(mock(MessageListener.class));
-		containerProps.setTransactionManager(mock(PlatformTransactionManager.class));
+		containerProps.setKafkaAwareTransactionManager(mock(KafkaAwareTransactionManager.class));
 		container = new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.start();
 		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.subBatchPerPartition"))

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -220,7 +220,7 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
 			factory.setConsumerFactory(consumerFactory());
 			factory.getContainerProperties().setAckMode(AckMode.BATCH);
-			factory.getContainerProperties().setTransactionManager(tm());
+			factory.getContainerProperties().setKafkaAwareTransactionManager(tm());
 			factory.setBatchListener(true);
 			factory.getContainerProperties().setSubBatchPerPartition(true);
 			return factory;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -206,7 +206,7 @@ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
 			factory.setConsumerFactory(consumerFactory());
 			factory.getContainerProperties().setAckMode(AckMode.BATCH);
-			factory.getContainerProperties().setTransactionManager(tm());
+			factory.getContainerProperties().setKafkaAwareTransactionManager(tm());
 			factory.getContainerProperties().setSubBatchPerPartition(true);
 			factory.setBatchListener(true);
 			return factory;