From 68f6730eef83500958faff6a0f564ac957c83762 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 15 Apr 2024 00:07:49 -0500 Subject: [PATCH] Fix flaky batch transaction tests Several batch transaction focused tests are randomly failing on CI builds. This commit aims to eliminate the failures by relaxing the test(s) to not require all input messages to arrive in a single invocation of a listener method. --- ...DefaultPulsarMessageListenerContainer.java | 4 +- ...ulsarMessageListenerContainerTxnTests.java | 122 ++++++++---------- .../listener/PulsarListenerTxnTests.java | 63 ++++----- .../pulsar/listener/PulsarTxnTestsBase.java | 13 -- 4 files changed, 91 insertions(+), 111 deletions(-) diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index 3d34e09a..fc690ac7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -565,7 +565,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess } }); } - catch (RuntimeException ex) { + catch (Throwable ex) { DefaultPulsarMessageListenerContainer.this.logger.error(ex, "Transaction rolled back"); } } @@ -651,7 +651,7 @@ public class DefaultPulsarMessageListenerContainer extends AbstractPulsarMess return this.transactionTemplate.execute(status -> doInvokeBatchListener(messages, messageList, inRetryMode, messagesPendingInBatch, getTransaction())); } - catch (RuntimeException e) { + catch (Throwable e) { DefaultPulsarMessageListenerContainer.this.logger.error(e, "Transaction rolled back"); return Collections.emptyList(); } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java index 9114a33f..2c30e440 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java @@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import java.time.Duration; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -39,6 +40,7 @@ import org.testcontainers.junit.jupiter.Testcontainers; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.test.support.PulsarConsumerTestUtil; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; @@ -58,7 +60,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { private PulsarClient client; - private PulsarTemplate pulsarTemplate; + private PulsarTemplate transactionalPulsarTemplate; private PulsarTransactionManager transactionManager; @@ -74,7 +76,8 @@ class DefaultPulsarMessageListenerContainerTxnTests { .serviceUrl(PULSAR_CONTAINER.getPulsarBrokerUrl()) .build(); var producerFactory = new DefaultPulsarProducerFactory(client); - pulsarTemplate = new PulsarTemplate<>(producerFactory); + transactionalPulsarTemplate = new PulsarTemplate<>(producerFactory); + transactionalPulsarTemplate.transactions().setEnabled(true); transactionManager = new PulsarTransactionManager(client); } @@ -90,8 +93,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { var containerProps = newContainerProps(); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); listenerLatch.countDown(); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, false, "msg1"); @@ -105,8 +107,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { var containerProps = newContainerProps(); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); listenerLatch.countDown(); }); @@ -122,8 +123,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { containerProps.setAckMode(AckMode.MANUAL); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarAcknowledgingMessageListener) (consumer, msg, ack) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); ack.acknowledge(); listenerLatch.countDown(); }); @@ -139,8 +139,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { containerProps.setAckMode(AckMode.MANUAL); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarAcknowledgingMessageListener) (consumer, msg, ack) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); ack.acknowledge(); PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); listenerLatch.countDown(); @@ -156,8 +155,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { var containerProps = newContainerProps(); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); listenerLatch.countDown(); throw new RuntimeException("BOOM"); }); @@ -172,8 +170,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { var containerProps = newContainerProps(); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out")); + transactionalPulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out")); listenerLatch.countDown(); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, false, "msg1"); @@ -187,8 +184,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { var containerProps = newContainerProps(); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out")); + transactionalPulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out")); PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); listenerLatch.countDown(); }); @@ -204,8 +200,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { var inputMsgs = List.of("msg1", "msg2", "msg3"); var listenerLatch = new CountDownLatch(inputMsgs.size()); containerProps.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); listenerLatch.countDown(); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, false, inputMsgs); @@ -221,8 +216,7 @@ class DefaultPulsarMessageListenerContainerTxnTests { var inputMsgs = List.of("msg1", "msg2", "msg3"); var listenerLatch = new CountDownLatch(inputMsgs.size()); containerProps.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - pulsarTemplate.transactions().setEnabled(true); - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); listenerLatch.countDown(); if (msg.getValue().equals("msg2")) { throw new RuntimeException("BOOM-msg2"); @@ -255,12 +249,12 @@ class DefaultPulsarMessageListenerContainerTxnTests { containerProps.setAckMode(AckMode.BATCH); containerProps.setSubscriptionType(SubscriptionType.Shared); var inputMsgs = List.of("msg1", "msg2", "msg3"); - var listenerLatch = new CountDownLatch(1); + var listenerLatch = new CountDownLatch(inputMsgs.size()); containerProps.setMessageListener((PulsarBatchMessageListener) (consumer, msgs) -> { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - pulsarTemplate.transactions().setEnabled(true); - msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out")); - listenerLatch.countDown(); + msgs.forEach((msg) -> { + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); + listenerLatch.countDown(); + }); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs); var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList(); @@ -278,12 +272,12 @@ class DefaultPulsarMessageListenerContainerTxnTests { containerProps.setAckMode(AckMode.BATCH); containerProps.setSubscriptionType(SubscriptionType.Exclusive); var inputMsgs = List.of("msg1", "msg2", "msg3"); - var listenerLatch = new CountDownLatch(1); + var listenerLatch = new CountDownLatch(inputMsgs.size()); containerProps.setMessageListener((PulsarBatchMessageListener) (consumer, msgs) -> { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - pulsarTemplate.transactions().setEnabled(true); - msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out")); - listenerLatch.countDown(); + msgs.forEach((msg) -> { + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); + listenerLatch.countDown(); + }); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs); var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList(); @@ -302,10 +296,8 @@ class DefaultPulsarMessageListenerContainerTxnTests { var inputMsgs = List.of("msg1", "msg2", "msg3"); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarBatchMessageListener) (consumer, msgs) -> { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - pulsarTemplate.transactions().setEnabled(true); - msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out")); - listenerLatch.countDown(); + msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out")); + CompletableFuture.runAsync(() -> listenerLatch.countDown()); throw new RuntimeException("NOPE"); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs); @@ -323,11 +315,9 @@ class DefaultPulsarMessageListenerContainerTxnTests { var inputMsgs = List.of("msg1", "msg2", "msg3"); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarBatchMessageListener) (consumer, msgs) -> { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - pulsarTemplate.transactions().setEnabled(true); - msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out")); - PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); + msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out")); listenerLatch.countDown(); + PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs); assertNoMessagesAvailableInOutputTopic(topicOut); @@ -343,22 +333,20 @@ class DefaultPulsarMessageListenerContainerTxnTests { var inputMsgs = List.of("msg1", "msg2", "msg3"); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarBatchMessageListener) (consumer, msgs) -> { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - pulsarTemplate.transactions().setEnabled(true); msgs.forEach((msg) -> { - if (msg.getValue().equals("msg2")) { - pulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out")); + if (msg.getValue().equals("msg1")) { + transactionalPulsarTemplate.executeInTransaction((t) -> t.send(topicOut, msg.getValue() + "-out")); } else { - pulsarTemplate.send(topicOut, msg.getValue() + "-out"); + transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out"); } }); - PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); listenerLatch.countDown(); + PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs); - // msg1 and msg2 get rollback but nested txn for msg2 gets committed - assertMessagesAvailableInOutputTopic(topicOut, "msg2-out"); + // msg2 and msg3 get rollback but nested txn for msg1 gets committed + assertMessagesAvailableInOutputTopic(topicOut, "msg1-out"); } @Test @@ -369,13 +357,11 @@ class DefaultPulsarMessageListenerContainerTxnTests { containerProps.setBatchListener(true); containerProps.setAckMode(AckMode.MANUAL); var inputMsgs = List.of("msg1", "msg2", "msg3"); - var listenerLatch = new CountDownLatch(1); + var listenerLatch = new CountDownLatch(inputMsgs.size()); containerProps.setMessageListener((PulsarBatchAcknowledgingMessageListener) (consumer, msgs, ack) -> { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - pulsarTemplate.transactions().setEnabled(true); - msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out")); + msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out")); ack.acknowledge(msgs.stream().map(Message::getMessageId).toList()); - listenerLatch.countDown(); + msgs.forEach((__) -> listenerLatch.countDown()); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs); var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList(); @@ -392,12 +378,10 @@ class DefaultPulsarMessageListenerContainerTxnTests { var inputMsgs = List.of("msg1", "msg2", "msg3"); var listenerLatch = new CountDownLatch(1); containerProps.setMessageListener((PulsarBatchAcknowledgingMessageListener) (consumer, msgs, ack) -> { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - pulsarTemplate.transactions().setEnabled(true); - msgs.forEach((msg) -> pulsarTemplate.send(topicOut, msg.getValue() + "-out")); + msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg.getValue() + "-out")); ack.acknowledge(msgs.stream().map(Message::getMessageId).toList()); - PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); listenerLatch.countDown(); + PulsarTransactionUtils.getResourceHolder(client).setRollbackOnly(); }); startContainerAndSendInputsThenWaitForLatch(topicIn, containerProps, listenerLatch, true, inputMsgs); assertNoMessagesAvailableInOutputTopic(topicOut); @@ -423,25 +407,29 @@ class DefaultPulsarMessageListenerContainerTxnTests { var container = new DefaultPulsarMessageListenerContainer<>(consumerFactory, containerProps); try { container.start(); - pulsarTemplate.transactions().setEnabled(false); - if (sendInBatch) { - inputMsgs.forEach((msg) -> pulsarTemplate.newMessage(msg) - .withTopic(topicIn) - .withProducerCustomizer((pb) -> pb.enableBatching(true) - .batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS) - .batchingMaxMessages(inputMsgs.size())) - .sendAsync()); - } - else { - inputMsgs.forEach((msg) -> pulsarTemplate.sendAsync(topicIn, msg)); - } + var nonTransactionalTemplate = newNonTransactionalTemplate(sendInBatch, inputMsgs.size()); + inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg)); assertThat(listenerLatch.await(sendInBatch ? 8 : 5, TimeUnit.SECONDS)).isTrue(); + if (sendInBatch) { + // Because the latch may fire before exception is thrown - give it a pause + Thread.sleep(500); + } } finally { container.stop(); } } + private PulsarTemplate newNonTransactionalTemplate(boolean sendInBatch, int numInBatch) { + List> customizers = List.of(); + if (sendInBatch) { + customizers = List.of((pb) -> pb.enableBatching(true) + .batchingMaxPublishDelay(2, TimeUnit.SECONDS) + .batchingMaxMessages(numInBatch)); + } + return new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(client, null, customizers)); + } + private void assertNoMessagesAvailableInOutputTopic(String topicOut) { assertThat(PulsarConsumerTestUtil.consumeMessages(client) .fromTopic(topicOut) 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 97ffa866..32e04947 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 @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; import java.util.Arrays; import java.util.List; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -33,6 +34,8 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.listener.PulsarListenerTxnTests.BatchListenerWithCommit.BatchListenerWithCommitConfig; import org.springframework.pulsar.listener.PulsarListenerTxnTests.BatchListenerWithRollback.BatchListenerWithRollbackConfig; @@ -51,10 +54,6 @@ 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) @@ -75,6 +74,16 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { .get()).map(Message::getValue).containsExactlyInAnyOrderElementsOf(expectedMessages); } + private PulsarTemplate newNonTransactionalTemplate(boolean sendInBatch, int numMessages) { + List> customizers = List.of(); + if (sendInBatch) { + customizers = List.of((pb) -> pb.enableBatching(true) + .batchingMaxPublishDelay(2, TimeUnit.SECONDS) + .batchingMaxMessages(numMessages)); + } + return new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient, null, customizers)); + } + @Nested @ContextConfiguration(classes = ListenerWithExternalTransactionConfig.class) class ListenerWithExternalTransaction { @@ -85,7 +94,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsCommitted() throws Exception { - sendInputMessageNonTransactionally(topicIn, "msg1"); + var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); + nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertMessagesAvailableInOutputTopic(topicOut, "msg1-out"); } @@ -118,7 +128,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsNotCommitted() throws Exception { - sendInputMessageNonTransactionally(topicIn, "msg1"); + var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); + nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertNoMessagesAvailableInOutputTopic(topicOut); } @@ -152,7 +163,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsCommitted() throws Exception { - sendInputMessageNonTransactionally(topicIn, "msg1"); + var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); + nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertMessagesAvailableInOutputTopic(topicOut, "msg1-out"); } @@ -184,7 +196,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @Test void producedMessageIsNotCommitted() throws Exception { - sendInputMessageNonTransactionally(topicIn, "msg1"); + var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); + nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertNoMessagesAvailableInOutputTopic(topicOut); } @@ -211,20 +224,16 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @ContextConfiguration(classes = BatchListenerWithCommitConfig.class) class BatchListenerWithCommit { - static final CountDownLatch latch = new CountDownLatch(1); static final String topicIn = "pltt-batch-lstnr-in"; static final String topicOut = "pltt-batch-lstnr-out"; static final List inputMsgs = List.of("msg1", "msg2", "msg3"); + static final CountDownLatch latch = new CountDownLatch(inputMsgs.size()); @Test void producedMessagesAreCommitted() throws Exception { - inputMsgs.forEach((msg) -> nonTransactionalPulsarTemplate.newMessage(msg) - .withTopic(topicIn) - .withProducerCustomizer((pb) -> pb.enableBatching(true) - .batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS) - .batchingMaxMessages(inputMsgs.size())) - .sendAsync()); - assertThat(latch.await(15, TimeUnit.SECONDS)).isTrue(); + var nonTransactionalTemplate = newNonTransactionalTemplate(true, inputMsgs.size()); + inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg)); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList(); assertMessagesAvailableInOutputTopic(topicOut, outputMsgs); } @@ -238,9 +247,10 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @PulsarListener(topics = topicIn, batch = true) void listen(List msgs) { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); - msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg + "-out")); - latch.countDown(); + msgs.forEach((msg) -> { + transactionalPulsarTemplate.send(topicOut, msg + "-out"); + latch.countDown(); + }); } } @@ -251,20 +261,16 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @ContextConfiguration(classes = BatchListenerWithRollbackConfig.class) class BatchListenerWithRollback { - static final CountDownLatch latch = new CountDownLatch(1); static final String topicIn = "pltt-batch-lstnr-rb-in"; static final String topicOut = "pltt-batch-lstnr-rb-out"; static final List inputMsgs = List.of("msg1", "msg2", "msg3"); + static final CountDownLatch latch = new CountDownLatch(1); @Test void producedMessagesAreNotCommitted() throws Exception { - inputMsgs.forEach((msg) -> nonTransactionalPulsarTemplate.newMessage(msg) - .withTopic(topicIn) - .withProducerCustomizer((pb) -> pb.enableBatching(true) - .batchingMaxPublishDelay(500, TimeUnit.MILLISECONDS) - .batchingMaxMessages(inputMsgs.size())) - .sendAsync()); - assertThat(latch.await(15, TimeUnit.SECONDS)).isTrue(); + var nonTransactionalTemplate = newNonTransactionalTemplate(true, inputMsgs.size()); + inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg)); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); assertNoMessagesAvailableInOutputTopic(topicOut); } @@ -277,9 +283,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { @PulsarListener(topics = topicIn, batch = true) void listen(List msgs) { - assertThat(msgs.size()).isEqualTo(inputMsgs.size()); msgs.forEach((msg) -> transactionalPulsarTemplate.send(topicOut, msg + "-out")); - latch.countDown(); + CompletableFuture.runAsync(() -> 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 d82e7906..c1f30fdd 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,7 +23,6 @@ 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; @@ -65,13 +64,8 @@ class PulsarTxnTestsBase { protected PulsarClient pulsarClient; @Autowired - @Qualifier("transactionalPulsarTemplate") protected PulsarTemplate transactionalPulsarTemplate; - @Autowired - @Qualifier("nonTransactionalPulsarTemplate") - protected PulsarTemplate nonTransactionalPulsarTemplate; - @Configuration(proxyBeanMethods = false) @EnablePulsar static class TopLevelConfig { @@ -96,13 +90,6 @@ class PulsarTxnTestsBase { 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) {