Merge pull request #42052 from vpavic

* pr/42052:
  Add support for configuring Pulsar client IO and listener threads

Closes gh-42052

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

@@ -44,6 +44,7 @@ import org.springframework.util.Assert;
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  * @since 3.2.0
  */
 @ConfigurationProperties("spring.pulsar")
@@ -136,6 +137,11 @@ public class PulsarProperties {
 		 */
 		private final Authentication authentication = new Authentication();
 
+		/**
+		 * Thread related configuration.
+		 */
+		private final Threads threads = new Threads();
+
 		/**
 		 * Failover settings.
 		 */
@@ -177,6 +183,10 @@ public class PulsarProperties {
 			return this.authentication;
 		}
 
+		public Threads getThreads() {
+			return this.threads;
+		}
+
 		public Failover getFailover() {
 			return this.failover;
 		}
@@ -959,6 +969,36 @@ public class PulsarProperties {
 
 	}
 
+	public static class Threads {
+
+		/**
+		 * Number of threads to be used for handling connections to brokers.
+		 */
+		private Integer io;
+
+		/**
+		 * Number of threads to be used for message listeners.
+		 */
+		private Integer listener;
+
+		public Integer getIo() {
+			return this.io;
+		}
+
+		public void setIo(Integer io) {
+			this.io = io;
+		}
+
+		public Integer getListener() {
+			return this.listener;
+		}
+
+		public void setListener(Integer listener) {
+			this.listener = listener;
+		}
+
+	}
+
 	public static class Failover {
 
 		/**

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

@@ -50,6 +50,7 @@ import org.springframework.util.StringUtils;
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 final class PulsarPropertiesMapper {
 
@@ -68,6 +69,8 @@ final class PulsarPropertiesMapper {
 		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.getThreads()::getIo).to(clientBuilder::ioThreads);
+		map.from(properties.getThreads()::getListener).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 @@ import static org.mockito.Mockito.never;
  * @author Chris Bono
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 class PulsarPropertiesMapperTests {
 
@@ -69,6 +70,8 @@ class PulsarPropertiesMapperTests {
 		properties.getClient().setConnectionTimeout(Duration.ofSeconds(1));
 		properties.getClient().setOperationTimeout(Duration.ofSeconds(2));
 		properties.getClient().setLookupTimeout(Duration.ofSeconds(3));
+		properties.getClient().getThreads().setIo(3);
+		properties.getClient().getThreads().setListener(10);
 		ClientBuilder builder = mock(ClientBuilder.class);
 		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
 				new PropertiesPulsarConnectionDetails(properties));
@@ -76,6 +79,8 @@ class PulsarPropertiesMapperTests {
 		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 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
  * @author Soby Chacko
  * @author Phillip Webb
  * @author Swamy Mavuri
+ * @author Vedran Pavic
  */
 class PulsarPropertiesTests {
 
@@ -88,6 +89,16 @@ class PulsarPropertiesTests {
 			assertThat(properties.getAuthentication().getParam()).containsEntry("token", "1234");
 		}
 
+		@Test
+		void bindThread() {
+			Map<String, String> map = new HashMap<>();
+			map.put("spring.pulsar.client.threads.io", "3");
+			map.put("spring.pulsar.client.threads.listener", "10");
+			PulsarProperties.Client properties = bindProperties(map).getClient();
+			assertThat(properties.getThreads().getIo()).isEqualTo(3);
+			assertThat(properties.getThreads().getListener()).isEqualTo(10);
+		}
+
 		@Test
 		void bindFailover() {
 			Map<String, String> map = new HashMap<>();