From f97be342ddea6b9dec4c0611869fadecc2552c88 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 22 Apr 2024 11:26:43 -0500 Subject: [PATCH] Add tests for @PulsarListener transactional attr See #661 --- .../listener/PulsarListenerTxnTests.java | 88 +++++++++++++++++++ .../pulsar/listener/PulsarTxnTestsBase.java | 21 ++++- 2 files changed, 105 insertions(+), 4 deletions(-) diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java index 32e04947..961ec4d7 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatException; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import java.time.Duration; import java.util.Arrays; @@ -31,9 +33,11 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.pulsar.core.PulsarTemplate; @@ -292,4 +296,88 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { } + @Nested + class TransactionsDisabledOnListener { + + static final String LISTENER_ID = "disabledOnListenerRequiredOnSettings"; + + @Test + void throwsExceptionWhenTransactionsAreRequired() { + assertThatIllegalStateException().isThrownBy(() -> { + var context = new AnnotationConfigApplicationContext(); + context.register(TopLevelConfig.class, TransactionsDisabledOnListenerConfig.class); + context.registerBean("containerPropsRequiredCustomizer", PulsarContainerPropertiesCustomizer.class, + () -> (c) -> c.transactions().setRequired(true)); + context.refresh(); + }).withMessage("Listener w/ id [%s] requested no transactions but txn are required".formatted(LISTENER_ID)); + } + + @Test + void disablesTransactionsWhenTransactionsAreNotRequired() { + try (var context = new AnnotationConfigApplicationContext()) { + context.register(TopLevelConfig.class, TransactionsDisabledOnListenerConfig.class); + context.registerBean("containerPropsNotRequiredCustomizer", PulsarContainerPropertiesCustomizer.class, + () -> (c) -> c.transactions().setRequired(false)); + context.refresh(); + var container = context.getBean(PulsarListenerEndpointRegistry.class).getListenerContainer(LISTENER_ID); + assertThat(container).isNotNull(); + assertThat(container.getContainerProperties()).satisfies((props) -> { + assertThat(props.transactions().isEnabled()).isFalse(); + assertThat(props.transactions().isRequired()).isFalse(); + }); + } + } + + static class TransactionsDisabledOnListenerConfig { + + @PulsarListener(id = LISTENER_ID, batch = true, transactional = "false", topics = "not-used") + void listen(List ignored) { + } + + } + + } + + @Nested + class TransactionsEnabledOnListener { + + static final String LISTENER_ID = "enabledOnListener"; + + @Test + void ignoresSettingWhenNoTxnManagerAvailable() { + assertThatException().isThrownBy(() -> { + var context = new AnnotationConfigApplicationContext(); + context.register(TopLevelConfig.class, TransactionsEnabledOnListenerConfig.class); + context.registerBean("removeTxnManagerCustomizer", PulsarContainerPropertiesCustomizer.class, + () -> (c) -> c.transactions().setTransactionManager(null)); + context.refresh(); + }) + .withCauseInstanceOf(IllegalStateException.class) + .havingCause() + .withMessage("Transactions are enabled but txn manager is not set"); + } + + @Test + void enablesTransactionsWhenTxnManagerAvailable() { + try (var context = new AnnotationConfigApplicationContext()) { + context.register(TopLevelConfig.class, TransactionsEnabledOnListenerConfig.class); + context.registerBean("containerPropsNotRequiredCustomizer", PulsarContainerPropertiesCustomizer.class, + () -> (c) -> c.transactions().setEnabled(false)); + context.refresh(); + var container = context.getBean(PulsarListenerEndpointRegistry.class).getListenerContainer(LISTENER_ID); + assertThat(container).isNotNull(); + assertThat(container.getContainerProperties().transactions().isEnabled()).isTrue(); + } + } + + static class TransactionsEnabledOnListenerConfig { + + @PulsarListener(id = LISTENER_ID, batch = true, transactional = "true", topics = "not-used") + void listen(List ignored) { + } + + } + + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java index c1f30fdd..31ccd9c3 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java @@ -98,14 +98,20 @@ class PulsarTxnTestsBase { } @Bean - PulsarListenerContainerFactory pulsarListenerContainerFactory( - PulsarConsumerFactory pulsarConsumerFactory, - PulsarAwareTransactionManager pulsarTransactionManager) { + PulsarContainerProperties pulsarContainerProperties(PulsarAwareTransactionManager pulsarTransactionManager) { var containerProps = new PulsarContainerProperties(); containerProps.transactions().setEnabled(true); containerProps.transactions().setRequired(false); containerProps.transactions().setTransactionManager(pulsarTransactionManager); - return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProps); + return containerProps; + } + + @Bean + PulsarListenerContainerFactory pulsarListenerContainerFactory( + PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProps, + ObjectProvider containerPropsCustomizer) { + containerPropsCustomizer.ifAvailable((c) -> c.customize(pulsarContainerProps)); + return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, pulsarContainerProps); } @Bean @@ -120,4 +126,11 @@ class PulsarTxnTestsBase { } + @FunctionalInterface + interface PulsarContainerPropertiesCustomizer { + + void customize(PulsarContainerProperties containerProperties); + + } + }