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

Clarify txn template in PulsarListenerTxnTests

Browse Source 
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.
This commit is contained in:
Chris Bono
2024-04-13 23:14:40 -05:00
parent 3e4c6bdc4e
commit c6f67842b9
2 changed files with 39 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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.<String>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<String> template;
private PulsarTemplate<String> 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<String> template;
private PulsarTemplate<String> 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<String> template;
private PulsarTemplate<String> 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<String> template;
private PulsarTemplate<String> 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<String> template;
private PulsarTemplate<String> transactionalPulsarTemplate;
@PulsarListener(topics = topicIn, batch = true)
void listen(List<String> 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<String> template;
private PulsarTemplate<String> transactionalPulsarTemplate;
@PulsarListener(topics = topicIn, batch = true)
void listen(List<String> 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");
}

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

@@ -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<String> pulsarTemplate;
@Qualifier("transactionalPulsarTemplate")
protected PulsarTemplate<String> transactionalPulsarTemplate;
@Autowired
@Qualifier("nonTransactionalPulsarTemplate")
protected PulsarTemplate<String> nonTransactionalPulsarTemplate;
@Configuration(proxyBeanMethods = false)
@EnablePulsar
@@ -84,12 +90,19 @@ class PulsarTxnTestsBase {
}
@Bean
PulsarTemplate<String> pulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
PulsarTemplate<String> transactionalPulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
var template = new PulsarTemplate<>(pulsarProducerFactory);
template.transactions().setEnabled(true);
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) {
@@ -102,6 +115,8 @@ class PulsarTxnTestsBase {
PulsarConsumerFactory<Object> pulsarConsumerFactory,
PulsarAwareTransactionManager pulsarTransactionManager) {
var containerProps = new PulsarContainerProperties();
containerProps.transactions().setEnabled(true);
containerProps.transactions().setRequired(false);
containerProps.transactions().setTransactionManager(pulsarTransactionManager);
return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProps);
}