From bfa22bcf8f63d4fab2b044436723a2a02e423e05 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Thu, 25 Aug 2022 17:18:18 -0400 Subject: [PATCH] Adding concurrency on PulsarListener --- .../pulsar/annotation/PulsarListener.java | 10 ++++ ...arListenerAnnotationBeanPostProcessor.java | 22 +++++++ ...ntPulsarMessageListenerContainerTests.java | 3 +- .../pulsar/listener/PulsarListenerTests.java | 58 +++++++++++++++++++ 4 files changed, 91 insertions(+), 2 deletions(-) diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java index bccd1408..93709d68 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java @@ -152,4 +152,14 @@ public @interface PulsarListener { */ String[] properties() default {}; + /** + * Override the container factory's {@code concurrency} setting for this listener. May + * be a property placeholder or SpEL expression that evaluates to a {@link Number}, in + * which case {@link Number#intValue()} is used to obtain the value. + *

+ * SpEL {@code #{...}} and property place holders {@code ${...}} are supported. + * @return the concurrency. + */ + String concurrency() default ""; + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java index 8e4b2716..56f980f8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java @@ -345,6 +345,11 @@ public class PulsarListenerAnnotationBeanPostProcessor endpoint.setSubscriptionType(getEndpointSubscriptionType(pulsarListener)); endpoint.setSchemaType(pulsarListener.schemaType()); + String concurrency = pulsarListener.concurrency(); + if (StringUtils.hasText(concurrency)) { + endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency")); + } + String autoStartup = pulsarListener.autoStartup(); if (StringUtils.hasText(autoStartup)) { endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup")); @@ -354,6 +359,23 @@ public class PulsarListenerAnnotationBeanPostProcessor endpoint.setBeanFactory(this.beanFactory); } + private Integer resolveExpressionAsInteger(String value, String attribute) { + Object resolved = resolveExpression(value); + Integer result = null; + if (resolved instanceof String) { + result = Integer.parseInt((String) resolved); + } + else if (resolved instanceof Number) { + result = ((Number) resolved).intValue(); + } + else if (resolved != null) { + throw new IllegalStateException( + THE_LEFT + attribute + "] must resolve to an Number or a String that can be parsed as an Integer. " + + RESOLVED_TO_LEFT + resolved.getClass() + RIGHT_FOR_LEFT + value + "]"); + } + return result; + } + private Boolean resolveExpressionAsBoolean(String value, String attribute) { Object resolved = resolveExpression(value); Boolean result = null; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java index 87753605..304fdc63 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/ConcurrentPulsarMessageListenerContainerTests.java @@ -32,13 +32,12 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; import org.junit.jupiter.api.Test; -import org.springframework.pulsar.core.AbstractContainerBaseTests; import org.springframework.pulsar.core.PulsarConsumerFactory; /** * @author Soby Chacko */ -public class ConcurrentPulsarMessageListenerContainerTests extends AbstractContainerBaseTests { +public class ConcurrentPulsarMessageListenerContainerTests { @Test @SuppressWarnings("unchecked") 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 e75be260..0e5b84d3 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 @@ -24,7 +24,9 @@ import java.util.Properties; import java.util.concurrent.CountDownLatch; 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.Test; import org.springframework.beans.factory.annotation.Autowired; @@ -54,6 +56,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; public class PulsarListenerTests extends AbstractContainerBaseTests { static CountDownLatch latch = new CountDownLatch(1); + static CountDownLatch latch1 = new CountDownLatch(3); @Autowired PulsarTemplate pulsarTemplate; @@ -61,8 +64,21 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { @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(); @@ -73,6 +89,42 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { 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 @EnablePulsar public static class Config { @@ -82,6 +134,12 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { latch.countDown(); } + @PulsarListener(id = "bar", topics = "concurrency-on-pl", subscriptionName = "subscription-2", + subscriptionType = "failover", concurrency = "3") + void listen2(String message) { + latch1.countDown(); + } + @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { Map config = new HashMap<>();