apply PulsarTxnTestsBase

Signed-off-by: Andrey Litvitski <[email protected]>

Author: therepanic

spring-pulsar/spring-pulsar.gradle

@@ -19,7 +19,6 @@ dependencies {
 	api 'org.springframework:spring-context'
 	api 'org.springframework:spring-messaging'
 	api 'org.springframework:spring-tx'
-	testImplementation libs.spring.boot.starter.pulsar
 	api (libs.spring.retry) {
 		exclude group: 'org.springframework'
 	}

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java

@@ -29,13 +29,11 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.pulsar.autoconfigure.PulsarAutoConfiguration;
-import org.springframework.boot.pulsar.autoconfigure.PulsarContainerFactoryCustomizer;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.listener.PulsarTxnTestsBase.TestPulsarContainerPropertiesCustomizer;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarListener;
-import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.listener.PulsarListenerTxnTests.BatchListenerWithCommit.BatchListenerWithCommitConfig;
@@ -276,10 +274,8 @@ void throwsExceptionWhenTransactionsAreRequired() {
 			assertThatIllegalStateException().isThrownBy(() -> {
 				var context = new AnnotationConfigApplicationContext();
 				context.register(TopLevelConfig.class, TransactionsDisabledOnListenerConfig.class);
-				context.registerBean(PulsarAutoConfiguration.class);
-				context.registerBean("containerPropsRequiredCustomizer", PulsarContainerFactoryCustomizer.class,
-						() -> ((PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>>) (
-								cf) -> cf.getContainerProperties().transactions().setRequired(true)));
+				context.registerBean("containerPropsRequiredCustomizer", TestPulsarContainerPropertiesCustomizer.class,
+						() -> containerProps -> containerProps.transactions().setRequired(true));
 				context.refresh();
 			}).withMessage("Listener w/ id [%s] requested no transactions but txn are required".formatted(LISTENER_ID));
 		}
@@ -288,10 +284,9 @@ void throwsExceptionWhenTransactionsAreRequired() {
 		void disablesTransactionsWhenTransactionsAreNotRequired() {
 			try (var context = new AnnotationConfigApplicationContext()) {
 				context.register(TopLevelConfig.class, TransactionsDisabledOnListenerConfig.class);
-				context.registerBean(PulsarAutoConfiguration.class);
-				context.registerBean("containerPropsNotRequiredCustomizer", PulsarContainerFactoryCustomizer.class,
-						() -> ((PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>>) (
-								cf) -> cf.getContainerProperties().transactions().setRequired(false)));
+				context.registerBean("containerPropsNotRequiredCustomizer",
+						TestPulsarContainerPropertiesCustomizer.class,
+						() -> containerProps -> containerProps.transactions().setRequired(false));
 				context.refresh();
 				var container = context.getBean(PulsarListenerEndpointRegistry.class).getListenerContainer(LISTENER_ID);
 				assertThat(container).isNotNull();
@@ -322,10 +317,8 @@ void ignoresSettingWhenNoTxnManagerAvailable() {
 			assertThatException().isThrownBy(() -> {
 				var context = new AnnotationConfigApplicationContext();
 				context.register(TopLevelConfig.class, TransactionsEnabledOnListenerConfig.class);
-				context.registerBean(PulsarAutoConfiguration.class);
-				context.registerBean("removeTxnManagerCustomizer", PulsarContainerFactoryCustomizer.class,
-						() -> ((PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>>) (
-								cf) -> cf.getContainerProperties().transactions().setTransactionManager(null)));
+				context.registerBean("removeTxnManagerCustomizer", TestPulsarContainerPropertiesCustomizer.class,
+						() -> containerProps -> containerProps.transactions().setTransactionManager(null));
 				context.refresh();
 			})
 				.withCauseInstanceOf(IllegalStateException.class)
@@ -337,10 +330,9 @@ void ignoresSettingWhenNoTxnManagerAvailable() {
 		void enablesTransactionsWhenTxnManagerAvailable() {
 			try (var context = new AnnotationConfigApplicationContext()) {
 				context.register(TopLevelConfig.class, TransactionsEnabledOnListenerConfig.class);
-				context.registerBean(PulsarAutoConfiguration.class);
-				context.registerBean("containerPropsNotRequiredCustomizer", PulsarContainerFactoryCustomizer.class,
-						() -> ((PulsarContainerFactoryCustomizer<ConcurrentPulsarListenerContainerFactory<?>>) (
-								cf) -> cf.getContainerProperties().transactions().setEnabled(false)));
+				context.registerBean("containerPropsNotRequiredCustomizer",
+						TestPulsarContainerPropertiesCustomizer.class,
+						() -> containerProps -> containerProps.transactions().setEnabled(false));
 				context.refresh();
 				var container = context.getBean(PulsarListenerEndpointRegistry.class).getListenerContainer(LISTENER_ID);
 				assertThat(container).isNotNull();

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java

@@ -0,0 +1,34 @@
+package org.springframework.pulsar.listener;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.pulsar.transaction.PulsarAwareTransactionManager;
+
+/**
+ * @author Chris Bono
+ * @author Andrey Litvitski
+ */
+public class PulsarTxnTestsBase {
+
+	@FunctionalInterface
+	public interface TestPulsarContainerPropertiesCustomizer {
+
+		void customize(PulsarContainerProperties containerProps);
+
+	}
+
+	@Bean
+	PulsarContainerProperties pulsarContainerProperties(PulsarAwareTransactionManager pulsarTransactionManager,
+			@Nullable TestPulsarContainerPropertiesCustomizer containerPropsCustomizer) {
+		var containerProps = new PulsarContainerProperties();
+		containerProps.transactions().setEnabled(true);
+		containerProps.transactions().setRequired(false);
+		containerProps.transactions().setTransactionManager(pulsarTransactionManager);
+		if (containerPropsCustomizer != null) {
+			containerPropsCustomizer.customize(containerProps);
+		}
+		return containerProps;
+	}
+
+}