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

Fix flaky batch transaction tests

Browse Source 
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.
This commit is contained in:
Chris Bono
2024-04-15 00:07:49 -05:00
parent 51c6b0143b
commit 68f6730eef
4 changed files with 91 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -565,7 +565,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
});
}
catch (RuntimeException ex) {
catch (Throwable ex) {
DefaultPulsarMessageListenerContainer.this.logger.error(ex, "Transaction rolled back");
}
}
@@ -651,7 +651,7 @@ public class DefaultPulsarMessageListenerContainer<T> 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();
}

122
spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java
View File

@@ -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<String> pulsarTemplate;
private PulsarTemplate<String> transactionalPulsarTemplate;
private PulsarTransactionManager transactionManager;
@@ -74,7 +76,8 @@ class DefaultPulsarMessageListenerContainerTxnTests {
.serviceUrl(PULSAR_CONTAINER.getPulsarBrokerUrl())
.build();
var producerFactory = new DefaultPulsarProducerFactory<String>(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<String> newNonTransactionalTemplate(boolean sendInBatch, int numInBatch) {
List<ProducerBuilderCustomizer<String>> 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.<String>consumeMessages(client)
.fromTopic(topicOut)

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

@@ -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.<String>consumeMessages(pulsarClient)
.fromTopic(topicOut)
@@ -75,6 +74,16 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase {
.get()).map(Message::getValue).containsExactlyInAnyOrderElementsOf(expectedMessages);
}
private PulsarTemplate<String> newNonTransactionalTemplate(boolean sendInBatch, int numMessages) {
List<ProducerBuilderCustomizer<String>> 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<String> 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<String> 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<String> 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<String> 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");
}

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

@@ -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<String> transactionalPulsarTemplate;
@Autowired
@Qualifier("nonTransactionalPulsarTemplate")
protected PulsarTemplate<String> nonTransactionalPulsarTemplate;
@Configuration(proxyBeanMethods = false)
@EnablePulsar
static class TopLevelConfig {
@@ -96,13 +90,6 @@ class PulsarTxnTestsBase {
return template;
}
@Bean
PulsarTemplate<String> nonTransactionalPulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
var template = new PulsarTemplate<>(pulsarProducerFactory);
template.transactions().setEnabled(false);
return template;
}
@Bean
public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
ObjectProvider<ConsumerBuilderCustomizer<String>> defaultConsumerCustomizersProvider) {