Apply transaction manager customizers to Pulsar

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 Shakil Ur Rehman
  * @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 transactionManager = new PulsarTransactionManager(pulsarClient);
+		transactionManagerCustomizers.ifAvailable((customizers) -> customizers.customize(transactionManager));
+		return transactionManager;
 	}
 
 	@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 Shakil Ur Rehman
  */
 class PulsarAutoConfigurationTests {
 
@@ -1114,6 +1118,30 @@ void whenTransactionEnabledTrueTransactionManagerIsAutoConfigured() {
 				.run((context) -> assertThat(context).hasSingleBean(PulsarAwareTransactionManager.class));
 		}
 
+		@Test
+		void transactionManagerWithTransactionExecutionListenersIsConfigured() {
+			TransactionExecutionListener listener = mock(TransactionExecutionListener.class);
+			this.contextRunner
+				.withConfiguration(AutoConfigurations.of(TransactionManagerCustomizationAutoConfiguration.class))
+				.withPropertyValues("spring.pulsar.transaction.enabled=true")
+				.withBean(TransactionExecutionListener.class, () -> listener)
+				.run((context) -> {
+					PulsarTransactionManager transactionManager = context.getBean(PulsarTransactionManager.class);
+					assertThat(transactionManager.getTransactionExecutionListeners()).containsExactly(listener);
+				});
+		}
+
+		@Test
+		void transactionManagerWithCustomizersIsConfigured() {
+			this.contextRunner
+				.withConfiguration(AutoConfigurations.of(TransactionManagerCustomizationAutoConfiguration.class))
+				.withPropertyValues("spring.pulsar.transaction.enabled=true", "spring.transaction.default-timeout=30s")
+				.run((context) -> {
+					PulsarTransactionManager transactionManager = context.getBean(PulsarTransactionManager.class);
+					assertThat(transactionManager.getDefaultTimeout()).isEqualTo(30);
+				});
+		}
+
 	}
 
 	record TestRecord() {