GH-3130 Switch to using 'setKafkaAwareTransactionManager' in KafkaMessageChannelBinder

olegz · olegz · commit 2916acfac774 · 2025-07-03T19:46:33.000+02:00
Resolves #3130
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -639,7 +639,7 @@ private <K, V> MessageProducer createConsumerEndpointCaptureHelper(
 		KafkaAwareTransactionManager<byte[], byte[]> transMan = transactionManager(
 				extendedConsumerProperties.getExtension().getTransactionManager());
 		if (transMan != null) {
-			containerProperties.setTransactionManager(transMan);
+			containerProperties.setKafkaAwareTransactionManager(transMan);
 		}
 		if (this.rebalanceListener != null) {
 			setupRebalanceListener(extendedConsumerProperties, containerProperties);

Original file line number	Diff line number	Diff line change
`@@ -639,7 +639,7 @@ private <K, V> MessageProducer createConsumerEndpointCaptureHelper(`
`639`	`639`	`KafkaAwareTransactionManager<byte[], byte[]> transMan = transactionManager(`
`640`	`640`	`extendedConsumerProperties.getExtension().getTransactionManager());`
`641`	`641`	`if (transMan != null) {`
`642`		`- containerProperties.setTransactionManager(transMan);`
	`642`	`+ containerProperties.setKafkaAwareTransactionManager(transMan);`
`643`	`643`	`}`
`644`	`644`	`if (this.rebalanceListener != null) {`
`645`	`645`	`setupRebalanceListener(extendedConsumerProperties, containerProperties);`