Merge pull request #42062 from vpavic

* pr/42062:
  Add support for configuring Pulsar listener container concurrency

Closes gh-42062

Author: philwebb

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarAutoConfiguration.java

@@ -69,6 +69,7 @@
  * @author Alexander Preuß
  * @author Phillip Webb
  * @author Jonas Geiregat
+ * @author Vedran Pavic
  * @since 3.2.0
  */
 @AutoConfiguration
@@ -187,7 +188,10 @@ ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
 		}
 		pulsarTransactionManager.ifUnique(containerProperties.transactions()::setTransactionManager);
 		this.propertiesMapper.customizeContainerProperties(containerProperties);
-		return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties);
+		ConcurrentPulsarListenerContainerFactory<Object> listenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>(
+				pulsarConsumerFactory, containerProperties);
+		this.propertiesMapper.customizeConcurrentPulsarListenerContainerFactory(listenerContainerFactory);
+		return listenerContainerFactory;
 	}
 
 	@Bean

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarProperties.java

@@ -811,6 +811,11 @@ public static class Listener {
 		 */
 		private SchemaType schemaType;
 
+		/**
+		 * Number of threads used by listener container.
+		 */
+		private Integer concurrency;
+
 		/**
 		 * Whether to record observations for when the Observations API is available and
 		 * the client supports it.
@@ -825,6 +830,14 @@ public void setSchemaType(SchemaType schemaType) {
 			this.schemaType = schemaType;
 		}
 
+		public Integer getConcurrency() {
+			return this.concurrency;
+		}
+
+		public void setConcurrency(Integer concurrency) {
+			this.concurrency = concurrency;
+		}
+
 		public boolean isObservationEnabled() {
 			return this.observationEnabled;
 		}

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesMapper.java

@@ -39,6 +39,7 @@
 
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.boot.json.JsonWriter;
+import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
 import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
@@ -198,6 +199,13 @@ private void customizePulsarContainerListenerProperties(PulsarContainerPropertie
 		map.from(properties::isObservationEnabled).to(containerProperties::setObservationEnabled);
 	}
 
+	<T> void customizeConcurrentPulsarListenerContainerFactory(
+			ConcurrentPulsarListenerContainerFactory<T> listenerContainerFactory) {
+		PulsarProperties.Listener properties = this.properties.getListener();
+		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		map.from(properties::getConcurrency).to(listenerContainerFactory::setConcurrency);
+	}
+
 	<T> void customizeReaderBuilder(ReaderBuilder<T> readerBuilder) {
 		PulsarProperties.Reader properties = this.properties.getReader();
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarReactivePropertiesMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2023 the original author or authors.
+ * Copyright 2012-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@
  *
  * @author Chris Bono
  * @author Phillip Webb
+ * @author Vedran Pavic
  */
 final class PulsarReactivePropertiesMapper {
 
@@ -93,6 +94,7 @@ private void customizePulsarContainerListenerProperties(ReactivePulsarContainerP
 		PulsarProperties.Listener properties = this.properties.getListener();
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
 		map.from(properties::getSchemaType).to(containerProperties::setSchemaType);
+		map.from(properties::getConcurrency).to(containerProperties::setConcurrency);
 	}
 
 	void customizeMessageReaderBuilder(ReactiveMessageReaderBuilder<?> builder) {

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesMapperTests.java

@@ -41,6 +41,7 @@
 
 import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Consumer;
 import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Failover.BackupCluster;
+import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarTemplate;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -272,6 +273,17 @@ void customizeContainerProperties() {
 		assertThat(containerProperties.transactions().isEnabled()).isTrue();
 	}
 
+	@Test
+	void customizeConcurrentPulsarListenerContainerFactory() {
+		PulsarProperties properties = new PulsarProperties();
+		properties.getListener().setConcurrency(10);
+		ConcurrentPulsarListenerContainerFactory<?> listenerContainerFactory = mock(
+				ConcurrentPulsarListenerContainerFactory.class);
+		new PulsarPropertiesMapper(properties)
+			.customizeConcurrentPulsarListenerContainerFactory(listenerContainerFactory);
+		then(listenerContainerFactory).should().setConcurrency(10);
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	void customizeReaderBuilder() {

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesTests.java

@@ -393,9 +393,11 @@ class ListenerProperties {
 		void bind() {
 			Map<String, String> map = new HashMap<>();
 			map.put("spring.pulsar.listener.schema-type", "avro");
+			map.put("spring.pulsar.listener.concurrency", "10");
 			map.put("spring.pulsar.listener.observation-enabled", "true");
 			PulsarProperties.Listener properties = bindProperties(map).getListener();
 			assertThat(properties.getSchemaType()).isEqualTo(SchemaType.AVRO);
+			assertThat(properties.getConcurrency()).isEqualTo(10);
 			assertThat(properties.isObservationEnabled()).isTrue();
 		}
 

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarReactivePropertiesMapperTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2023 the original author or authors.
+ * Copyright 2012-2024 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,6 +48,7 @@
  *
  * @author Chris Bono
  * @author Phillip Webb
+ * @author Vedran Pavic
  */
 class PulsarReactivePropertiesMapperTests {
 
@@ -120,10 +121,12 @@ void customizeContainerProperties() {
 		PulsarProperties properties = new PulsarProperties();
 		properties.getConsumer().getSubscription().setType(SubscriptionType.Shared);
 		properties.getListener().setSchemaType(SchemaType.AVRO);
+		properties.getListener().setConcurrency(10);
 		ReactivePulsarContainerProperties<Object> containerProperties = new ReactivePulsarContainerProperties<>();
 		new PulsarReactivePropertiesMapper(properties).customizeContainerProperties(containerProperties);
 		assertThat(containerProperties.getSubscriptionType()).isEqualTo(SubscriptionType.Shared);
 		assertThat(containerProperties.getSchemaType()).isEqualTo(SchemaType.AVRO);
+		assertThat(containerProperties.getConcurrency()).isEqualTo(10);
 	}
 
 	@Test