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

Failover consumer test improvements

Browse Source 
Ensure that all 3 failover consumers participate in the subscription
This commit is contained in:
Soby Chacko
2023-02-01 12:31:17 -05:00
parent 4db9357dac
commit c288e9e496
1 changed files with 23 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java
View File

@@ -60,21 +60,32 @@ class FailoverConsumerTests implements PulsarTestContainerSupport {
.build();
DefaultPulsarConsumerFactory<String> pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient,
config);
CountDownLatch latch = new CountDownLatch(3);
CountDownLatch latch1 = new CountDownLatch(1);
CountDownLatch latch2 = new CountDownLatch(1);
CountDownLatch latch3 = new CountDownLatch(1);
PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties
.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> latch.countDown());
.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> latch1.countDown());
pulsarContainerProperties.setSubscriptionType(SubscriptionType.Failover);
pulsarContainerProperties.setSchema(Schema.STRING);
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container.start();
DefaultPulsarMessageListenerContainer<String> container1 = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container1.start();
pulsarContainerProperties
.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> latch2.countDown());
DefaultPulsarMessageListenerContainer<String> container2 = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container2.start();
pulsarContainerProperties
.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> latch3.countDown());
DefaultPulsarMessageListenerContainer<String> container3 = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container3.start();
Map<String, Object> prodConfig = Map.of("topicName", "my-part-topic-1", "messageRoutingMode",
MessageRoutingMode.CustomPartition);
DefaultPulsarProducerFactory<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
@@ -88,8 +99,13 @@ class FailoverConsumerTests implements PulsarTestContainerSupport {
pulsarTemplate.newMessage("hello buzz doe")
.withProducerCustomizer(builder -> builder.messageRouter(new BuzzRouter())).sendAsync();
boolean await = latch.await(10, TimeUnit.SECONDS);
assertThat(await).isTrue();
boolean await1 = latch1.await(10, TimeUnit.SECONDS);
boolean await2 = latch2.await(10, TimeUnit.SECONDS);
boolean await3 = latch3.await(10, TimeUnit.SECONDS);
assertThat(await1).isTrue();
assertThat(await2).isTrue();
assertThat(await3).isTrue();
}
static class FooRouter implements MessageRouter {