Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add tests for @PulsarListener transactional attr

Browse Source 
See #661
This commit is contained in:
Chris Bono
2024-04-22 11:26:43 -05:00
parent 02e45fd0cc
commit f97be342dd
2 changed files with 105 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String> 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<String> ignored) {
}
}
}
}

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

@@ -98,14 +98,20 @@ class PulsarTxnTestsBase {
}
@Bean
PulsarListenerContainerFactory pulsarListenerContainerFactory(
PulsarConsumerFactory<Object> 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<Object> pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProps,
ObjectProvider<PulsarContainerPropertiesCustomizer> 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);
}
}