Port "Use builder to autoconfigure PulsarClient (#394)" from 0.2.x (#396)

- Replace PulsarClientFactoryBean with PulsarClientFactory

spring-pulsar-docs/src/main/asciidoc/pulsar.adoc

@@ -24,8 +24,8 @@ TIP: The value must be a valid {apache-pulsar-docs}/client-libraries-java/#conne
 
 You can further configure the client by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application properties.
 
-NOTE: If you are not using the starter, you will need to configure and register the `PulsarClientFactoryBean` yourself.
-It has a constructor that accepts a map of Pulsar https://pulsar.apache.org/docs/2.11.x/client-libraries-java/#client[native properties].
+NOTE: If you are not using the starter, you will need to configure and register the `PulsarClient` yourself.
+There is a `DefaultPulsarClientFactory` that accepts a builder customizer that can be used to help with this.
 
 [[client-authentication]]
 === Authentication

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

@@ -23,7 +23,6 @@ import java.time.Duration;
 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.DeadLetterPolicy;
 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.Bean;
 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 class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 		}
 
 		@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/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java

@@ -1,65 +0,0 @@
-/*
- * Copyright 2022-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.config;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
-
-import org.apache.pulsar.client.api.PulsarClient;
-
-import org.springframework.beans.factory.FactoryBean;
-import org.springframework.beans.factory.config.AbstractFactoryBean;
-import org.springframework.core.log.LogAccessor;
-import org.springframework.lang.Nullable;
-
-/**
- * {@link FactoryBean} implementation for the {@link PulsarClient}.
- *
- * @author Soby Chacko
- * @author Chris Bono
- */
-public class PulsarClientFactoryBean extends AbstractFactoryBean<PulsarClient> {
-
-	private final LogAccessor logger = new LogAccessor(this.getClass());
-
-	private final Map<String, Object> config = new HashMap<>();
-
-	public PulsarClientFactoryBean(Map<String, Object> config) {
-		Objects.requireNonNull(config, "Config map cannot be null");
-		this.config.putAll(config);
-	}
-
-	@Override
-	public Class<?> getObjectType() {
-		return PulsarClient.class;
-	}
-
-	@Override
-	protected PulsarClient createInstance() throws Exception {
-		return PulsarClient.builder().loadConf(this.config).build();
-	}
-
-	@Override
-	protected void destroyInstance(@Nullable PulsarClient instance) throws Exception {
-		if (instance != null) {
-			this.logger.info(() -> "Closing client " + instance);
-			instance.close();
-		}
-	}
-
-}

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2022-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.core;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+import org.springframework.util.Assert;
+
+/**
+ * Default implementation for {@link PulsarClientFactory}.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+public class DefaultPulsarClientFactory implements PulsarClientFactory {
+
+	private final PulsarClientBuilderCustomizer customizer;
+
+	/**
+	 * Construct a factory that creates clients using a default Pulsar client builder with
+	 * no modifications other than the specified service url.
+	 * @param serviceUrl the service url
+	 */
+	public DefaultPulsarClientFactory(String serviceUrl) {
+		this((clientBuilder -> clientBuilder.serviceUrl(serviceUrl)));
+	}
+
+	/**
+	 * Construct a factory that creates clients using a customized Pulsar client builder.
+	 * @param customizer the customizer to apply to the builder
+	 */
+	public DefaultPulsarClientFactory(PulsarClientBuilderCustomizer customizer) {
+		Assert.notNull(customizer, "customizer must not be null");
+		this.customizer = customizer;
+	}
+
+	@Override
+	public PulsarClient createClient() throws PulsarClientException {
+		var clientBuilder = PulsarClient.builder();
+		this.customizer.customize(clientBuilder);
+		return clientBuilder.build();
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientBuilderCustomizer.java

@@ -0,0 +1,35 @@
+/*
+ * 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.core;
+
+import org.apache.pulsar.client.api.ClientBuilder;
+
+/**
+ * The interface to customize a {@link ClientBuilder}.
+ *
+ * @author Chris Bono
+ */
+@FunctionalInterface
+public interface PulsarClientBuilderCustomizer {
+
+	/**
+	 * Customizes a {@link ClientBuilder}.
+	 * @param clientBuilder the builder to customize
+	 */
+	void customize(ClientBuilder clientBuilder);
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientFactory.java

@@ -0,0 +1,37 @@
+/*
+ * 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.core;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+
+/**
+ * Pulsar client factory interface.
+ *
+ * @author Soby Chacko
+ * @author Chris Bono
+ */
+public interface PulsarClientFactory {
+
+	/**
+	 * Create a client.
+	 * @return the created client instance
+	 * @throws PulsarClientException if an error occurs creating the client
+	 */
+	PulsarClient createClient() throws PulsarClientException;
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java

@@ -0,0 +1,63 @@
+/*
+ * 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.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatRuntimeException;
+
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link DefaultPulsarClientFactory}.
+ *
+ * @author Chris Bono
+ */
+class DefaultPulsarClientFactoryTests {
+
+	@Test
+	void constructWithServiceUrl() throws PulsarClientException {
+		var clientFactory = new DefaultPulsarClientFactory("pulsar://localhost:5150");
+		assertThat(clientFactory.createClient()).hasFieldOrPropertyWithValue("conf.serviceUrl",
+				"pulsar://localhost:5150");
+	}
+
+	@Test
+	void constructWithCustomizer() throws PulsarClientException {
+		var clientFactory = new DefaultPulsarClientFactory(
+				(clientBuilder) -> clientBuilder.serviceUrl("pulsar://localhost:5150"));
+		assertThat(clientFactory.createClient()).hasFieldOrPropertyWithValue("conf.serviceUrl",
+				"pulsar://localhost:5150");
+	}
+
+	@Test
+	void constructWithNullCustomizer() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> new DefaultPulsarClientFactory((PulsarClientBuilderCustomizer) null))
+				.withMessage("customizer must not be null");
+	}
+
+	@Test
+	void customizerThrowsException() {
+		var clientFactory = new DefaultPulsarClientFactory((clientBuilder) -> {
+			throw new RuntimeException("Who turned out the lights?");
+		});
+		assertThatRuntimeException().isThrownBy(clientFactory::createClient).withMessage("Who turned out the lights?");
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java

@@ -39,6 +39,7 @@ import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Messages;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.RedeliveryBackoff;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -61,10 +62,10 @@ import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarListener;
 import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
-import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
 import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
 import org.springframework.pulsar.core.ConsumerBuilderCustomizer;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
@@ -109,8 +110,8 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
 		}
 
 		@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/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java

@@ -35,8 +35,8 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarListener;
 import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
-import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
@@ -135,8 +135,8 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul
 		}
 
 		@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/src/test/java/org/springframework/pulsar/observation/ObservationTests.java

@@ -23,7 +23,6 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.Deque;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -40,8 +39,8 @@ import org.springframework.lang.Nullable;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarListener;
 import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
-import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.config.PulsarListenerContainerFactory;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
 import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
@@ -175,8 +174,8 @@ public class ObservationTests implements PulsarTestContainerSupport {
 		}
 
 		@Bean
-		PulsarClientFactoryBean pulsarClientFactoryBean() {
-			return new PulsarClientFactoryBean(Map.of("serviceUrl", PulsarTestContainerSupport.getPulsarBrokerUrl()));
+		public PulsarClient pulsarClient() throws PulsarClientException {
+			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
 		}
 
 		@Bean(name = "observationTestsTemplate")

spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java

@@ -39,8 +39,8 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.pulsar.annotation.EnablePulsar;
 import org.springframework.pulsar.annotation.PulsarReader;
 import org.springframework.pulsar.config.DefaultPulsarReaderContainerFactory;
-import org.springframework.pulsar.config.PulsarClientFactoryBean;
 import org.springframework.pulsar.config.PulsarReaderContainerFactory;
+import org.springframework.pulsar.core.DefaultPulsarClientFactory;
 import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
 import org.springframework.pulsar.core.DefaultPulsarReaderFactory;
 import org.springframework.pulsar.core.PulsarProducerFactory;
@@ -78,8 +78,8 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
 		}
 
 		@Bean
-		public PulsarClientFactoryBean pulsarClientFactoryBean() {
-			return new PulsarClientFactoryBean(Map.of("serviceUrl", PulsarTestContainerSupport.getPulsarBrokerUrl()));
+		public PulsarClient pulsarClient() throws PulsarClientException {
+			return new DefaultPulsarClientFactory(PulsarTestContainerSupport.getPulsarBrokerUrl()).createClient();
 		}
 
 		@Bean