From c6f67842b9fc5fee23da20cf9140af4e3f336d76 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sat, 13 Apr 2024 23:14:40 -0500 Subject: [PATCH] Clarify txn template in PulsarListenerTxnTests The PulsarListenerTxnTests is currently flaky. A possible cause is the fact that the tests use the same template to send the input and output messages and they just flip the transactional flag back and forth. This commit configures a txn and non txn template for the test. --- .../listener/PulsarListenerTxnTests.java | 52 ++++++++----------- .../pulsar/listener/PulsarTxnTestsBase.java | 19 ++++++- 2 files changed, 39 insertions(+), 32 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 f257147f..97ffa866 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 @@ -51,6 +51,10 @@ import org.springframework.transaction.annotation.Transactional; */ class PulsarListenerTxnTests extends PulsarTxnTestsBase { + private void sendInputMessageNonTransactionally(String topic, String msg) { + nonTransactionalPulsarTemplate.send(topic, msg); + } + private void assertNoMessagesAvailableInOutputTopic(String topicOut) { assertThat(PulsarConsumerTestUtil.consumeMessages(pulsarClient) .fromTopic(topicOut) @@ -81,8 +85,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsCommitted() throws Exception { - pulsarTemplate.transactions().setEnabled(false); - pulsarTemplate.sendAsync(topicIn, "msg1"); + sendInputMessageNonTransactionally(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertMessagesAvailableInOutputTopic(topicOut, "msg1-out"); } @@ -92,13 +95,12 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { static class ListenerWithExternalTransactionConfig { @Autowired - private PulsarTemplate template; + private PulsarTemplate transactionalPulsarTemplate; @Transactional @PulsarListener(topics = topicIn, ackMode = AckMode.RECORD) void listen(String msg) { - template.transactions().setEnabled(true); - template.send(topicOut, msg + "-out"); + transactionalPulsarTemplate.send(topicOut, msg + "-out"); latch.countDown(); } @@ -116,8 +118,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsNotCommitted() throws Exception { - pulsarTemplate.transactions().setEnabled(false); - pulsarTemplate.sendAsync(topicIn, "msg1"); + sendInputMessageNonTransactionally(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertNoMessagesAvailableInOutputTopic(topicOut); } @@ -127,13 +128,12 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { static class ListenerWithExternalTransactionRollbackConfig { @Autowired - private PulsarTemplate template; + private PulsarTemplate transactionalPulsarTemplate; @Transactional @PulsarListener(topics = topicIn, ackMode = AckMode.RECORD) void listen(String msg) { - template.transactions().setEnabled(true); - template.send(topicOut, msg + "-out"); + transactionalPulsarTemplate.send(topicOut, msg + "-out"); latch.countDown(); throw new RuntimeException("BOOM"); } @@ -152,8 +152,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsCommitted() throws Exception { - pulsarTemplate.transactions().setEnabled(false); - pulsarTemplate.sendAsync(topicIn, "msg1"); + sendInputMessageNonTransactionally(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertMessagesAvailableInOutputTopic(topicOut, "msg1-out"); } @@ -163,12 +162,11 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { static class RecordListenerWithCommitConfig { @Autowired - private PulsarTemplate template; + private PulsarTemplate transactionalPulsarTemplate; @PulsarListener(topics = topicIn, ackMode = AckMode.RECORD) void listen(String msg) { - template.transactions().setEnabled(true); - template.send(topicOut, msg + "-out"); + transactionalPulsarTemplate.send(topicOut, msg + "-out"); latch.countDown(); } @@ -186,8 +184,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsNotCommitted() throws Exception { - pulsarTemplate.transactions().setEnabled(false); - pulsarTemplate.sendAsync(topicIn, "msg1"); + sendInputMessageNonTransactionally(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertNoMessagesAvailableInOutputTopic(topicOut); } @@ -197,12 +194,11 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { static class RecordListenerWithRollbackConfig { @Autowired - private PulsarTemplate template; + private PulsarTemplate transactionalPulsarTemplate; @PulsarListener(topics = topicIn, ackMode = AckMode.RECORD) void listen(String msg) { - template.transactions().setEnabled(true); - template.send(topicOut, msg + "-out"); + transactionalPulsarTemplate.send(topicOut, msg + "-out"); latch.countDown(); throw new RuntimeException("BOOM-record"); } @@ -222,8 +218,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessagesAreCommitted() throws Exception { - pulsarTemplate.transactions().setEnabled(false); - inputMsgs.forEach((msg) -> pulsarTemplate.newMessage(msg) + inputMsgs.forEach((msg) -> nonTransactionalPulsarTemplate.newMessage(msg) .withTopic(topicIn) .withProducerCustomizer((pb) -> pb.enableBatching(true) .batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS) @@ -239,13 +234,12 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { static class BatchListenerWithCommitConfig { @Autowired - private PulsarTemplate template; + private PulsarTemplate transactionalPulsarTemplate; @PulsarListener(topics = topicIn, batch = true) void listen(List msgs) { assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - template.transactions().setEnabled(true); - msgs.forEach((msg) -> template.send(topicOut, msg + "-out")); + msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg + "-out")); latch.countDown(); } @@ -264,8 +258,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessagesAreNotCommitted() throws Exception { - pulsarTemplate.transactions().setEnabled(false); - inputMsgs.forEach((msg) -> pulsarTemplate.newMessage(msg) + inputMsgs.forEach((msg) -> nonTransactionalPulsarTemplate.newMessage(msg) .withTopic(topicIn) .withProducerCustomizer((pb) -> pb.enableBatching(true) .batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS) @@ -280,13 +273,12 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { static class BatchListenerWithRollbackConfig { @Autowired - private PulsarTemplate template; + private PulsarTemplate transactionalPulsarTemplate; @PulsarListener(topics = topicIn, batch = true) void listen(List msgs) { assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - template.transactions().setEnabled(true); - msgs.forEach((msg) -> template.send(topicOut, msg + "-out")); + msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg + "-out")); latch.countDown(); throw new RuntimeException("BOOM-batch"); } 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 77d016b7..d82e7906 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 @@ -23,6 +23,7 @@ import org.testcontainers.junit.jupiter.Testcontainers; import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; @@ -64,7 +65,12 @@ class PulsarTxnTestsBase { protected PulsarClient pulsarClient; @Autowired - protected PulsarTemplate pulsarTemplate; + @Qualifier("transactionalPulsarTemplate") + protected PulsarTemplate transactionalPulsarTemplate; + + @Autowired + @Qualifier("nonTransactionalPulsarTemplate") + protected PulsarTemplate nonTransactionalPulsarTemplate; @Configuration(proxyBeanMethods = false) @EnablePulsar @@ -84,12 +90,19 @@ class PulsarTxnTestsBase { } @Bean - PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { + PulsarTemplate transactionalPulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { var template = new PulsarTemplate<>(pulsarProducerFactory); template.transactions().setEnabled(true); return template; } + @Bean + PulsarTemplate nonTransactionalPulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { + var template = new PulsarTemplate<>(pulsarProducerFactory); + template.transactions().setEnabled(false); + return template; + } + @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, ObjectProvider> defaultConsumerCustomizersProvider) { @@ -102,6 +115,8 @@ class PulsarTxnTestsBase { PulsarConsumerFactory pulsarConsumerFactory, PulsarAwareTransactionManager pulsarTransactionManager) { var containerProps = new PulsarContainerProperties(); + containerProps.transactions().setEnabled(true); + containerProps.transactions().setRequired(false); containerProps.transactions().setTransactionManager(pulsarTransactionManager); return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProps); }