Apply TransactionManagerCustomizers consistently for Pulsar

Signed-off-by: Yanming Zhou <zhouyanming@gmail.com>

Author: quaff

module/spring-boot-pulsar/src/main/java/org/springframework/boot/pulsar/autoconfigure/PulsarAutoConfiguration.java

@@ -30,6 +30,7 @@
 import org.springframework.boot.pulsar.autoconfigure.PulsarProperties.Defaults.SchemaInfo;
 import org.springframework.boot.pulsar.autoconfigure.PulsarProperties.Defaults.TypeMapping;
 import org.springframework.boot.thread.Threading;
+import org.springframework.boot.transaction.autoconfigure.TransactionManagerCustomizers;
 import org.springframework.boot.util.LambdaSafe;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -80,6 +81,7 @@
  * @author Alexander Preuß
  * @author Phillip Webb
  * @author Jonas Geiregat
+ * @author Yanming Zhou
  * @since 4.0.0
  */
 @AutoConfiguration
@@ -303,8 +305,11 @@ DefaultPulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
 	@Bean
 	@ConditionalOnMissingBean(PulsarAwareTransactionManager.class)
 	@ConditionalOnBooleanProperty("spring.pulsar.transaction.enabled")
-	PulsarTransactionManager pulsarTransactionManager(PulsarClient pulsarClient) {
-		return new PulsarTransactionManager(pulsarClient);
+	PulsarTransactionManager pulsarTransactionManager(PulsarClient pulsarClient,
+			ObjectProvider<TransactionManagerCustomizers> transactionManagerCustomizers) {
+		PulsarTransactionManager pulsarTransactionManager = new PulsarTransactionManager(pulsarClient);
+		transactionManagerCustomizers.ifAvailable((customizers) -> customizers.customize(pulsarTransactionManager));
+		return pulsarTransactionManager;
 	}
 
 	@SuppressWarnings("unchecked")

module/spring-boot-pulsar/src/test/java/org/springframework/boot/pulsar/autoconfigure/PulsarAutoConfigurationTests.java

@@ -40,6 +40,7 @@
 import org.springframework.boot.test.context.FilteredClassLoader;
 import org.springframework.boot.test.context.TestConfiguration;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.boot.transaction.autoconfigure.TransactionManagerCustomizationAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.annotation.Order;
@@ -78,7 +79,9 @@
 import org.springframework.pulsar.function.PulsarFunctionAdministration;
 import org.springframework.pulsar.listener.PulsarContainerProperties.TransactionSettings;
 import org.springframework.pulsar.transaction.PulsarAwareTransactionManager;
+import org.springframework.pulsar.transaction.PulsarTransactionManager;
 import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.TransactionExecutionListener;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.BDDMockito.given;
@@ -92,6 +95,7 @@
  * @author Alexander Preuß
  * @author Soby Chacko
  * @author Phillip Webb
+ * @author Yanming Zhou
  */
 class PulsarAutoConfigurationTests {
 
@@ -173,6 +177,18 @@ void whenHasUserDefinedContainerFactoryCustomizersBeanDoesNotAutoConfigureBean()
 				.isSameAs(customizers));
 	}
 
+	@Test
+	void pulsarTransactionManagerWithTransactionExecutionListenersIsConfigured() {
+		TransactionExecutionListener listener = mock(TransactionExecutionListener.class);
+		this.contextRunner.withPropertyValues("spring.pulsar.transaction.enabled=true")
+			.withConfiguration(AutoConfigurations.of(TransactionManagerCustomizationAutoConfiguration.class))
+			.withBean(TransactionExecutionListener.class, () -> listener)
+			.run((context) -> {
+				PulsarTransactionManager transactionManager = context.getBean(PulsarTransactionManager.class);
+				assertThat(transactionManager.getTransactionExecutionListeners()).containsExactly(listener);
+			});
+	}
+
 	@Nested
 	class ClientTests {