Refactor PulsarClient auto-configuration

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java

@@ -23,7 +23,6 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
@@ -36,6 +35,7 @@
 import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -57,7 +57,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.pulsar.annotation.EnablePulsar;
-import org.springframework.pulsar.config.PulsarClientFactoryBean;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.DefaultTopicResolver;
@@ -111,8 +111,8 @@ public PulsarProducerFactory<String> pulsarProducerFactory(PulsarClient pulsarCl
 		}
 
 		@Bean
-		public PulsarClientFactoryBean pulsarClientFactoryBean() {
-			return new PulsarClientFactoryBean(Map.of("serviceUrl", PulsarTestContainerSupport.getPulsarBrokerUrl()));
+		public PulsarClient pulsarClient() throws PulsarClientException {
+			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
 		}
 
 		@Bean

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

@@ -19,6 +19,7 @@
 import java.util.Optional;
 
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
 
 import org.springframework.beans.factory.ObjectProvider;
@@ -30,14 +31,15 @@
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
-import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.core.CachingPulsarProducerFactory;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.DefaultTopicResolver;
 import org.springframework.pulsar.core.PulsarAdministration;
+import org.springframework.pulsar.core.PulsarClientBuilderCustomizer;
 import org.springframework.pulsar.core.PulsarConsumerFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
 import org.springframework.pulsar.core.PulsarReaderFactory;
@@ -71,8 +73,21 @@ public PulsarAutoConfiguration(PulsarProperties properties) {
 
 	@Bean
 	@ConditionalOnMissingBean
-	public PulsarClientFactoryBean pulsarClientFactoryBean() {
-		return new PulsarClientFactoryBean(this.properties.buildClientProperties());
+	public PulsarClientBuilderConfigurer pulsarClientBuilderConfigurer(PulsarProperties pulsarProperties,
+			ObjectProvider<PulsarClientBuilderCustomizer> customizers) {
+		return new PulsarClientBuilderConfigurer(pulsarProperties, customizers.orderedStream().toList());
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public PulsarClient pulsarClient(PulsarClientBuilderConfigurer configurer) {
+		var clientFactory = new DefaultPulsarClientFactory(configurer::configure);
+		try {
+			return clientFactory.createClient();
+		}
+		catch (PulsarClientException ex) {
+			throw new IllegalArgumentException("Failed to create client: " + ex.getMessage(), ex);
+		}
 	}
 
 	@Bean

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

@@ -0,0 +1,172 @@
+/*
+ * Copyright 2023-2023 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.autoconfigure;
+
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.pulsar.client.api.ClientBuilder;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SizeUnit;
+
+import org.springframework.boot.context.properties.PropertyMapper;
+import org.springframework.boot.util.LambdaSafe;
+import org.springframework.pulsar.autoconfigure.PulsarProperties.Client;
+import org.springframework.pulsar.core.PulsarClientBuilderCustomizer;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+import org.springframework.util.unit.DataSize;
+
+/**
+ * Configure Pulsar {@link ClientBuilder} with sensible defaults and apply a list of
+ * optional {@link PulsarClientBuilderCustomizer customizers}.
+ *
+ * @author Chris Bono
+ */
+public class PulsarClientBuilderConfigurer {
+
+	private final PulsarProperties properties;
+
+	private final List<PulsarClientBuilderCustomizer> customizers;
+
+	/**
+	 * Creates a new configurer that will use the given properties for configuration.
+	 * @param properties properties to use
+	 * @param customizers list of customizers to apply or empty list if no customizers
+	 */
+	public PulsarClientBuilderConfigurer(PulsarProperties properties, List<PulsarClientBuilderCustomizer> customizers) {
+		Assert.notNull(properties, "properties must not be null");
+		Assert.notNull(customizers, "customizers must not be null");
+		this.properties = properties;
+		this.customizers = customizers;
+	}
+
+	/**
+	 * Configure the specified {@link ClientBuilder}. The builder can be further tuned and
+	 * default settings can be overridden.
+	 * @param clientBuilder the {@link ClientBuilder} instance to configure
+	 */
+	public void configure(ClientBuilder clientBuilder) {
+		applyProperties(this.properties, clientBuilder);
+		applyCustomizers(this.customizers, clientBuilder);
+	}
+
+	@SuppressWarnings("deprecation")
+	protected void applyProperties(PulsarProperties pulsarProperties, ClientBuilder clientBuilder) {
+		var map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		var clientProperties = pulsarProperties.getClient();
+		map.from(clientProperties::getServiceUrl).to(clientBuilder::serviceUrl);
+		map.from(clientProperties::getListenerName).to(clientBuilder::listenerName);
+		map.from(clientProperties::getNumIoThreads).to(clientBuilder::ioThreads);
+		map.from(clientProperties::getNumListenerThreads).to(clientBuilder::listenerThreads);
+		map.from(clientProperties::getNumConnectionsPerBroker).to(clientBuilder::connectionsPerBroker);
+		map.from(clientProperties::getMaxConcurrentLookupRequest).to(clientBuilder::maxConcurrentLookupRequests);
+		map.from(clientProperties::getMaxLookupRequest).to(clientBuilder::maxLookupRequests);
+		map.from(clientProperties::getMaxLookupRedirects).to(clientBuilder::maxLookupRedirects);
+		map.from(clientProperties::getMaxNumberOfRejectedRequestPerConnection)
+				.to(clientBuilder::maxNumberOfRejectedRequestPerConnection);
+		map.from(clientProperties::getUseTcpNoDelay).to(clientBuilder::enableTcpNoDelay);
+
+		// Authentication properties
+		applyAuthentication(clientProperties, clientBuilder);
+
+		// TLS properties
+		map.from(clientProperties::getUseTls).to(clientBuilder::enableTls);
+		map.from(clientProperties::getTlsHostnameVerificationEnable).to(clientBuilder::enableTlsHostnameVerification);
+		map.from(clientProperties::getTlsTrustCertsFilePath).to(clientBuilder::tlsTrustCertsFilePath);
+		map.from(clientProperties::getTlsCertificateFilePath).to(clientBuilder::tlsCertificateFilePath);
+		map.from(clientProperties::getTlsKeyFilePath).to(clientBuilder::tlsKeyFilePath);
+		map.from(clientProperties::getTlsAllowInsecureConnection).to(clientBuilder::allowTlsInsecureConnection);
+		map.from(clientProperties::getUseKeyStoreTls).to(clientBuilder::useKeyStoreTls);
+		map.from(clientProperties::getSslProvider).to(clientBuilder::sslProvider);
+		map.from(clientProperties::getTlsTrustStoreType).to(clientBuilder::tlsTrustStoreType);
+		map.from(clientProperties::getTlsTrustStorePath).to(clientBuilder::tlsTrustStorePath);
+		map.from(clientProperties::getTlsTrustStorePassword).to(clientBuilder::tlsTrustStorePassword);
+		map.from(clientProperties::getTlsCiphers).to(clientBuilder::tlsCiphers);
+		map.from(clientProperties::getTlsProtocols).to(clientBuilder::tlsProtocols);
+
+		map.from(clientProperties::getStatsInterval).as(Duration::toSeconds).to(clientBuilder,
+				(cb, val) -> cb.statsInterval(val, TimeUnit.SECONDS));
+		map.from(clientProperties::getKeepAliveInterval).asInt(Duration::toMillis).to(clientBuilder,
+				(cb, val) -> cb.keepAliveInterval(val, TimeUnit.MILLISECONDS));
+		map.from(clientProperties::getConnectionTimeout).asInt(Duration::toMillis).to(clientBuilder,
+				(cb, val) -> cb.connectionTimeout(val, TimeUnit.MILLISECONDS));
+		map.from(clientProperties::getOperationTimeout).asInt(Duration::toMillis).to(clientBuilder,
+				(cb, val) -> cb.operationTimeout(val, TimeUnit.MILLISECONDS));
+		map.from(clientProperties::getLookupTimeout).asInt(Duration::toMillis).to(clientBuilder,
+				(cb, val) -> cb.lookupTimeout(val, TimeUnit.MILLISECONDS));
+		map.from(clientProperties::getInitialBackoffInterval).as(Duration::toMillis).to(clientBuilder,
+				(cb, val) -> cb.startingBackoffInterval(val, TimeUnit.MILLISECONDS));
+		map.from(clientProperties::getMaxBackoffInterval).as(Duration::toMillis).to(clientBuilder,
+				(cb, val) -> cb.maxBackoffInterval(val, TimeUnit.MILLISECONDS));
+		map.from(clientProperties::getEnableBusyWait).to(clientBuilder::enableBusyWait);
+		map.from(clientProperties::getMemoryLimit).as(DataSize::toBytes).to(clientBuilder,
+				(cb, val) -> cb.memoryLimit(val, SizeUnit.BYTES));
+		map.from(clientProperties::getEnableTransaction).to(clientBuilder::enableTransaction);
+		map.from(clientProperties::getProxyServiceUrl)
+				.to((proxyUrl) -> clientBuilder.proxyServiceUrl(proxyUrl, clientProperties.getProxyProtocol()));
+		map.from(clientProperties::getDnsLookupBindAddress)
+				.to((bindAddr) -> clientBuilder.dnsLookupBind(bindAddr, clientProperties.getDnsLookupBindPort()));
+		map.from(clientProperties::getSocks5ProxyAddress).as(this::parseSocketAddress)
+				.to(clientBuilder::socks5ProxyAddress);
+		map.from(clientProperties::getSocks5ProxyUsername).to(clientBuilder::socks5ProxyUsername);
+		map.from(clientProperties::getSocks5ProxyPassword).to(clientBuilder::socks5ProxyPassword);
+	}
+
+	private void applyAuthentication(Client clientProperties, ClientBuilder clientBuilder) {
+		if (StringUtils.hasText(clientProperties.getAuthParams())
+				&& !CollectionUtils.isEmpty(clientProperties.getAuthentication())) {
+			throw new IllegalArgumentException(
+					"Cannot set both spring.pulsar.client.authParams and spring.pulsar.client.authentication.*");
+		}
+		var authPluginClass = clientProperties.getAuthPluginClassName();
+		if (StringUtils.hasText(authPluginClass)) {
+			var authParams = clientProperties.getAuthParams();
+			if (clientProperties.getAuthentication() != null) {
+				authParams = AuthParameterUtils.maybeConvertToEncodedParamString(clientProperties.getAuthentication());
+			}
+			try {
+				clientBuilder.authentication(authPluginClass, authParams);
+			}
+			catch (PulsarClientException.UnsupportedAuthenticationException ex) {
+				throw new IllegalArgumentException("Unable to configure authentication: " + ex.getMessage(), ex);
+			}
+		}
+	}
+
+	private InetSocketAddress parseSocketAddress(String address) {
+		try {
+			var uri = URI.create(address);
+			return new InetSocketAddress(uri.getHost(), uri.getPort());
+		}
+		catch (Exception e) {
+			throw new IllegalArgumentException("Invalid address: " + e.getMessage(), e);
+		}
+	}
+
+	protected void applyCustomizers(List<PulsarClientBuilderCustomizer> clientBuilderCustomizers,
+			ClientBuilder clientBuilder) {
+		LambdaSafe.callbacks(PulsarClientBuilderCustomizer.class, clientBuilderCustomizers, clientBuilder)
+				.withLogger(PulsarClientBuilderConfigurer.class)
+				.invoke((customizer) -> customizer.customize(clientBuilder));
+	}
+
+}

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

@@ -110,10 +110,6 @@ public Map<String, Object> buildConsumerProperties() {
 		return new HashMap<>(this.consumer.buildProperties());
 	}
 
-	public Map<String, Object> buildClientProperties() {
-		return new HashMap<>(this.client.buildProperties());
-	}
-
 	public Map<String, Object> buildProducerProperties() {
 		return new HashMap<>(this.producer.buildProperties());
 	}
@@ -773,67 +769,6 @@ public void setSocks5ProxyPassword(String socks5ProxyPassword) {
 			this.socks5ProxyPassword = socks5ProxyPassword;
 		}
 
-		public Map<String, Object> buildProperties() {
-			if (StringUtils.hasText(this.getAuthParams()) && !CollectionUtils.isEmpty(this.getAuthentication())) {
-				throw new IllegalArgumentException(
-						"Cannot set both spring.pulsar.client.authParams and spring.pulsar.client.authentication.*");
-			}
-
-			PulsarProperties.Properties properties = new Properties();
-
-			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-			map.from(this::getServiceUrl).to(properties.in("serviceUrl"));
-			map.from(this::getListenerName).to(properties.in("listenerName"));
-			map.from(this::getAuthPluginClassName).to(properties.in("authPluginClassName"));
-			map.from(this::getAuthParams).to(properties.in("authParams"));
-			map.from(this::getAuthentication).as(AuthParameterUtils::maybeConvertToEncodedParamString)
-					.to(properties.in("authParams"));
-			map.from(this::getOperationTimeout).as(Duration::toMillis).to(properties.in("operationTimeoutMs"));
-			map.from(this::getLookupTimeout).as(Duration::toMillis).to(properties.in("lookupTimeoutMs"));
-			map.from(this::getNumIoThreads).to(properties.in("numIoThreads"));
-			map.from(this::getNumListenerThreads).to(properties.in("numListenerThreads"));
-			map.from(this::getNumConnectionsPerBroker).to(properties.in("connectionsPerBroker"));
-			map.from(this::getUseTcpNoDelay).to(properties.in("useTcpNoDelay"));
-			map.from(this::getUseTls).to(properties.in("useTls"));
-			map.from(this::getTlsHostnameVerificationEnable).to(properties.in("tlsHostnameVerificationEnable"));
-			map.from(this::getTlsTrustCertsFilePath).to(properties.in("tlsTrustCertsFilePath"));
-			map.from(this::getTlsCertificateFilePath).to(properties.in("tlsCertificateFilePath"));
-			map.from(this::getTlsKeyFilePath).to(properties.in("tlsKeyFilePath"));
-			map.from(this::getTlsAllowInsecureConnection).to(properties.in("tlsAllowInsecureConnection"));
-			map.from(this::getUseKeyStoreTls).to(properties.in("useKeyStoreTls"));
-			map.from(this::getSslProvider).to(properties.in("sslProvider"));
-			map.from(this::getTlsTrustStoreType).to(properties.in("tlsTrustStoreType"));
-			map.from(this::getTlsTrustStorePath).to(properties.in("tlsTrustStorePath"));
-			map.from(this::getTlsTrustStorePassword).to(properties.in("tlsTrustStorePassword"));
-			map.from(this::getTlsCiphers).to(properties.in("tlsCiphers"));
-			map.from(this::getTlsProtocols).to(properties.in("tlsProtocols"));
-			map.from(this::getStatsInterval).as(Duration::toSeconds).to(properties.in("statsIntervalSeconds"));
-			map.from(this::getMaxConcurrentLookupRequest).to(properties.in("concurrentLookupRequest"));
-			map.from(this::getMaxLookupRequest).to(properties.in("maxLookupRequest"));
-			map.from(this::getMaxLookupRedirects).to(properties.in("maxLookupRedirects"));
-			map.from(this::getMaxNumberOfRejectedRequestPerConnection)
-					.to(properties.in("maxNumberOfRejectedRequestPerConnection"));
-			map.from(this::getKeepAliveInterval).asInt(Duration::toSeconds)
-					.to(properties.in("keepAliveIntervalSeconds"));
-			map.from(this::getConnectionTimeout).asInt(Duration::toMillis).to(properties.in("connectionTimeoutMs"));
-			map.from(this::getRequestTimeout).asInt(Duration::toMillis).to(properties.in("requestTimeoutMs"));
-			map.from(this::getInitialBackoffInterval).as(Duration::toNanos)
-					.to(properties.in("initialBackoffIntervalNanos"));
-			map.from(this::getMaxBackoffInterval).as(Duration::toNanos).to(properties.in("maxBackoffIntervalNanos"));
-			map.from(this::getEnableBusyWait).to(properties.in("enableBusyWait"));
-			map.from(this::getMemoryLimit).as(DataSize::toBytes).to(properties.in("memoryLimitBytes"));
-			map.from(this::getProxyServiceUrl).to(properties.in("proxyServiceUrl"));
-			map.from(this::getProxyProtocol).to(properties.in("proxyProtocol"));
-			map.from(this::getEnableTransaction).to(properties.in("enableTransaction"));
-			map.from(this::getDnsLookupBindAddress).to(properties.in("dnsLookupBindAddress"));
-			map.from(this::getDnsLookupBindPort).to(properties.in("dnsLookupBindPort"));
-			map.from(this::getSocks5ProxyAddress).to(properties.in("socks5ProxyAddress"));
-			map.from(this::getSocks5ProxyUsername).to(properties.in("socks5ProxyUsername"));
-			map.from(this::getSocks5ProxyPassword).to(properties.in("socks5ProxyPassword"));
-
-			return properties;
-		}
-
 	}
 
 	public static class Function {