diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index 5dfc6299..f13d907b 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -249,6 +249,75 @@ The following message listener types are available when using Spring for Apache We will see the details about these various message listeners in the sections below. +Before doing so however, lets take a closer look at the container itself + +===== DefaultPulsarMessageListenerContainer + +This is a single consumer based message listener container. +Here is it's constructor. + +==== +[source, java] +---- +public DefaultPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, + PulsarContainerProperties pulsarContainerProperties) +} +---- +==== + +It receives a `PulsarConsumerFactory` that it uses to create the consumer and a `PulsarContainerProperties` object that contains information about the container properties. +`PulsarContainerProperties` has the following constructors. + +==== +[source, java] +---- +public PulsarContainerProperties(String... topics) + +public PulsarContainerProperties(Pattern topicPattern) +---- +==== + +You can provide the topic information through `PulsarContainerProperties` or as a consumer property that is provided to the consumer factory. +Here is an example of using the `DefaultPulsarMessageListenerContainer`. + +==== +[source, java] +---- +Map config = new HashMap<>(); +config.put("topics", "my-topic"); +PulsarConsumerFactory pulsarConsumerFactorY = DefaultPulsarConsumerFactory<>(pulsarClient, config); + +PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); + +pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { + }); + +DefaultPulsarMessageListenerContainer pulsarListenerContainer = new DefaultPulsarMessageListenerContainer(pulsarConsumerFacotyr, + pulsarContainerProperties); + +return pulsarListenerContainer; +---- +==== + +`DefaultPulsarMessageListenerContainer` only creates a single consumer. +If you want to have multiple consumers managed through multiple threads, you need to use `ConcurrentPulsarMessageListenerContainer`. + +===== ConcurrentPulsarMessageListenerContainer + +`ConcurrentPulsarMessageListenerContainer` has the following constructor. + +==== +[source, java] +---- +public ConcurrentPulsarMessageListenerContainer(PulsarConsumerFactory pulsarConsumerFactory, + PulsarContainerProperties pulsarContainerProperties) +---- +==== + +`ConcurrentPulsarMessageListenerContainer` allows to specify a `concurrency` property through a setter. +Concurrency of more than `1` is only allowed on non-exclusive subscriptions (`failover`, `shared` and `key-shared`). +You can only have the default `1` for concurrency when you have an exclusive subscription mode. + ==== Consuming the Records In this section, we are going to see how the message listener container enables both single record and batch based message consumption. diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 0e5b84d3..0550dfb8 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -17,6 +17,7 @@ package org.springframework.pulsar.listener; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.HashMap; import java.util.Map; @@ -26,10 +27,11 @@ import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; -import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; @@ -42,10 +44,13 @@ import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.core.AbstractContainerBaseTests; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.PulsarTopic; import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** @@ -61,84 +66,12 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { @Autowired PulsarTemplate pulsarTemplate; - @Autowired - private PulsarListenerEndpointRegistry registry; - @Autowired private PulsarClient pulsarClient; - @BeforeAll - static void setup() throws Exception { - PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build(); - - String topicName = "persistent://public/default/concurrency-on-pl"; - int numPartitions = 3; - admin.topics().createPartitionedTopic(topicName, numPartitions); - } - - @Test - void testPulsarListenerProvidedConsumerProperties() throws Exception { - - final PulsarContainerProperties pulsarContainerProperties = this.registry.getListenerContainer("foo") - .getContainerProperties(); - final Properties pulsarConsumerProperties = pulsarContainerProperties.getPulsarConsumerProperties(); - assertThat(pulsarConsumerProperties.size()).isEqualTo(2); - assertThat(pulsarConsumerProperties.get("topicNames")).isEqualTo("foo-1"); - assertThat(pulsarConsumerProperties.get("subscriptionName")).isEqualTo("subscription-1"); - pulsarTemplate.send("hello foo"); - assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); - } - - @Test - void concurrencyOnPulsarListenerWithFailoverSubscription() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Map.of("batchingEnabled", false)); - PulsarTemplate customTemplate = new PulsarTemplate<>(pulsarProducerFactory); - - final ConcurrentPulsarMessageListenerContainer bar = (ConcurrentPulsarMessageListenerContainer) this.registry - .getListenerContainer("bar"); - - assertThat(bar.getConcurrency()).isEqualTo(3); - - customTemplate.sendAsync("concurrency-on-pl", "hello john doe"); - customTemplate.sendAsync("concurrency-on-pl", "hello alice doe"); - customTemplate.sendAsync("concurrency-on-pl", "hello buzz doe"); - - assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); - } - - @Test - void nonDefaultConcurrencySettingNotAllowedOnExclusiveSubscriptions() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Map.of("batchingEnabled", false)); - PulsarTemplate customTemplate = new PulsarTemplate<>(pulsarProducerFactory); - - final ConcurrentPulsarMessageListenerContainer bar = (ConcurrentPulsarMessageListenerContainer) this.registry - .getListenerContainer("bar"); - - assertThat(bar.getConcurrency()).isEqualTo(3); - - customTemplate.sendAsync("concurrency-on-pl", "hello john doe"); - customTemplate.sendAsync("concurrency-on-pl", "hello alice doe"); - customTemplate.sendAsync("concurrency-on-pl", "hello buzz doe"); - - assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); - } - - @Configuration + @Configuration(proxyBeanMethods = false) @EnablePulsar - public static class Config { - - @PulsarListener(id = "foo", properties = { "subscriptionName=subscription-1", "topicNames=foo-1" }) - void listen1(String message) { - latch.countDown(); - } - - @PulsarListener(id = "bar", topics = "concurrency-on-pl", subscriptionName = "subscription-2", - subscriptionType = "failover", concurrency = "3") - void listen2(String message) { - latch1.countDown(); - } + public static class TopLevelConfig { @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { @@ -176,6 +109,114 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { return pulsarListenerContainerFactory; } + @Bean + PulsarAdministration pulsarAdministration() { + return new PulsarAdministration(PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl())); + } + + @Bean + PulsarTopic partitionedTopic() { + return PulsarTopic.builder("persistent://public/default/concurrency-on-pl").numberOfPartitions(3).build(); + } + + } + + @Nested + @ContextConfiguration(classes = PulsarListenerBasicTestCases.TestPulsarListenersForBasicScenario.class) + class PulsarListenerBasicTestCases { + + @Test + void testPulsarListenerProvidedConsumerProperties(@Autowired PulsarListenerEndpointRegistry registry) + throws Exception { + + final PulsarContainerProperties pulsarContainerProperties = registry.getListenerContainer("foo") + .getContainerProperties(); + final Properties pulsarConsumerProperties = pulsarContainerProperties.getPulsarConsumerProperties(); + assertThat(pulsarConsumerProperties.size()).isEqualTo(2); + assertThat(pulsarConsumerProperties.get("topicNames")).isEqualTo("foo-1"); + assertThat(pulsarConsumerProperties.get("subscriptionName")).isEqualTo("subscription-1"); + pulsarTemplate.send("hello foo"); + assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void concurrencyOnPulsarListenerWithFailoverSubscription(@Autowired PulsarListenerEndpointRegistry registry) + throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Map.of("batchingEnabled", false)); + PulsarTemplate customTemplate = new PulsarTemplate<>(pulsarProducerFactory); + + final ConcurrentPulsarMessageListenerContainer bar = (ConcurrentPulsarMessageListenerContainer) registry + .getListenerContainer("bar"); + + assertThat(bar.getConcurrency()).isEqualTo(3); + + customTemplate.sendAsync("concurrency-on-pl", "hello john doe"); + customTemplate.sendAsync("concurrency-on-pl", "hello alice doe"); + customTemplate.sendAsync("concurrency-on-pl", "hello buzz doe"); + + assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void nonDefaultConcurrencySettingNotAllowedOnExclusiveSubscriptions( + @Autowired PulsarListenerEndpointRegistry registry) throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Map.of("batchingEnabled", false)); + PulsarTemplate customTemplate = new PulsarTemplate<>(pulsarProducerFactory); + + final ConcurrentPulsarMessageListenerContainer bar = (ConcurrentPulsarMessageListenerContainer) registry + .getListenerContainer("bar"); + + assertThat(bar.getConcurrency()).isEqualTo(3); + + customTemplate.sendAsync("concurrency-on-pl", "hello john doe"); + customTemplate.sendAsync("concurrency-on-pl", "hello alice doe"); + customTemplate.sendAsync("concurrency-on-pl", "hello buzz doe"); + + assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class TestPulsarListenersForBasicScenario { + + @PulsarListener(id = "foo", properties = { "subscriptionName=subscription-1", "topicNames=foo-1" }) + void listen1(String message) { + latch.countDown(); + } + + @PulsarListener(id = "bar", topics = "concurrency-on-pl", subscriptionName = "subscription-2", + subscriptionType = "failover", concurrency = "3") + void listen2(String message) { + latch1.countDown(); + } + + } + + } + + @Nested + class NegativeConcurrency { + + @Test + void exclusiveSubscriptionNotAllowedToHaveMultipleConsumers() { + assertThatThrownBy( + () -> new AnnotationConfigApplicationContext(TopLevelConfig.class, ConcurrencyConfig.class)) + .rootCause().isInstanceOf(IllegalStateException.class) + .hasMessage("concurrency > 1 is not allowed on Exclusive subscription type"); + } + + @EnablePulsar + static class ConcurrencyConfig { + + @PulsarListener(id = "foobar", topics = "concurrency-on-pl", subscriptionName = "subscription-3", + concurrency = "3") + void listen3(String message) { + } + + } + } }