Adding concurrency on PulsarListener

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.
+	 * <p>
+	 * SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
+	 * @return the concurrency.
+	 */
+	String concurrency() default "";
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java

@@ -345,6 +345,11 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
 		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<K, V>
 		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;

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")

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<String> 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<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+				Map.of("batchingEnabled", false));
+		PulsarTemplate<String> 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<String> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
+				Map.of("batchingEnabled", false));
+		PulsarTemplate<String> 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<String> pulsarProducerFactory(PulsarClient pulsarClient) {
 			Map<String, Object> config = new HashMap<>();