Add support for configuring Pulsar client IO and listener threads

This commit adds configuration properties that allow users to configure
number of IO threads and listener threads used by the Pulsar client.

Author: vpavic

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")
@@ -131,6 +132,16 @@ public static class Client {
 		 */
 		private Duration connectionTimeout = Duration.ofSeconds(10);
 
+		/**
+		 * Number of threads to be used for handling connections to brokers.
+		 */
+		private Integer ioThreads;
+
+		/**
+		 * Number of threads to be used for message listeners.
+		 */
+		private Integer listenerThreads;
+
 		/**
 		 * Authentication settings.
 		 */
@@ -173,6 +184,22 @@ public void setConnectionTimeout(Duration connectionTimeout) {
 			this.connectionTimeout = connectionTimeout;
 		}
 
+		public Integer getIoThreads() {
+			return this.ioThreads;
+		}
+
+		public void setIoThreads(Integer ioThreads) {
+			this.ioThreads = ioThreads;
+		}
+
+		public Integer getListenerThreads() {
+			return this.listenerThreads;
+		}
+
+		public void setListenerThreads(Integer listenerThreads) {
+			this.listenerThreads = listenerThreads;
+		}
+
 		public Authentication getAuthentication() {
 			return this.authentication;
 		}

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

@@ -50,6 +50,7 @@
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 final class PulsarPropertiesMapper {
 
@@ -68,6 +69,8 @@ void customizeClientBuilder(ClientBuilder clientBuilder, PulsarConnectionDetails
 		map.from(properties::getConnectionTimeout).to(timeoutProperty(clientBuilder::connectionTimeout));
 		map.from(properties::getOperationTimeout).to(timeoutProperty(clientBuilder::operationTimeout));
 		map.from(properties::getLookupTimeout).to(timeoutProperty(clientBuilder::lookupTimeout));
+		map.from(properties::getIoThreads).to(clientBuilder::ioThreads);
+		map.from(properties::getListenerThreads).to(clientBuilder::listenerThreads);
 		map.from(this.properties.getTransaction()::isEnabled).whenTrue().to(clientBuilder::enableTransaction);
 		customizeAuthentication(properties.getAuthentication(), clientBuilder::authentication);
 		customizeServiceUrlProviderBuilder(clientBuilder::serviceUrl, clientBuilder::serviceUrlProvider, properties,

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

@@ -59,6 +59,7 @@
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 class PulsarPropertiesMapperTests {
 
@@ -69,13 +70,17 @@ void customizeClientBuilderWhenHasNoAuthentication() {
 		properties.getClient().setConnectionTimeout(Duration.ofSeconds(1));
 		properties.getClient().setOperationTimeout(Duration.ofSeconds(2));
 		properties.getClient().setLookupTimeout(Duration.ofSeconds(3));
+		properties.getClient().setIoThreads(3);
+		properties.getClient().setListenerThreads(10);
 		ClientBuilder builder = mock(ClientBuilder.class);
 		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
 				new PropertiesPulsarConnectionDetails(properties));
 		then(builder).should().serviceUrl("https://example.com");
 		then(builder).should().connectionTimeout(1000, TimeUnit.MILLISECONDS);
 		then(builder).should().operationTimeout(2000, TimeUnit.MILLISECONDS);
 		then(builder).should().lookupTimeout(3000, TimeUnit.MILLISECONDS);
+		then(builder).should().ioThreads(3);
+		then(builder).should().listenerThreads(10);
 	}
 
 	@Test

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 {
 
@@ -71,11 +72,15 @@ void bind() {
 			map.put("spring.pulsar.client.operation-timeout", "1s");
 			map.put("spring.pulsar.client.lookup-timeout", "2s");
 			map.put("spring.pulsar.client.connection-timeout", "12s");
+			map.put("spring.pulsar.client.io-threads", "3");
+			map.put("spring.pulsar.client.listener-threads", "10");
 			PulsarProperties.Client properties = bindProperties(map).getClient();
 			assertThat(properties.getServiceUrl()).isEqualTo("my-service-url");
 			assertThat(properties.getOperationTimeout()).isEqualTo(Duration.ofMillis(1000));
 			assertThat(properties.getLookupTimeout()).isEqualTo(Duration.ofMillis(2000));
 			assertThat(properties.getConnectionTimeout()).isEqualTo(Duration.ofMillis(12000));
+			assertThat(properties.getIoThreads()).isEqualTo(3);
+			assertThat(properties.getListenerThreads()).isEqualTo(10);
 		}
 
 		@Test