Add support for configuring Pulsar listener container concurrency

This commit adds configuration property that allows users to configure
Pulsar message listener container concurrency.

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

@@ -44,6 +44,7 @@
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  * @since 3.2.0
  */
 @ConfigurationProperties("spring.pulsar")
@@ -801,6 +802,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.
@@ -815,6 +821,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;
@@ -50,6 +51,7 @@
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 final class PulsarPropertiesMapper {
 
@@ -195,6 +197,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/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;
@@ -59,6 +60,7 @@
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 class PulsarPropertiesMapperTests {
 
@@ -267,6 +269,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

@@ -54,6 +54,7 @@
  * @author Soby Chacko
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 class PulsarPropertiesTests {
 
@@ -382,9 +383,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();
 		}