From 82f5bcbaf361b25fc559f120364199728528e32e Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sat, 29 Apr 2023 11:04:22 -0500 Subject: [PATCH] Port "Use builder to autoconfigure PulsarClient (#394)" from 0.2.x (#396) - Replace PulsarClientFactoryBean with PulsarClientFactory --- .../src/main/asciidoc/pulsar.adoc | 4 +- .../listener/ReactivePulsarListenerTests.java | 8 +-- .../config/PulsarClientFactoryBean.java | 65 ------------------- .../core/DefaultPulsarClientFactory.java | 59 +++++++++++++++++ .../core/PulsarClientBuilderCustomizer.java | 35 ++++++++++ .../pulsar/core/PulsarClientFactory.java | 37 +++++++++++ .../core/DefaultPulsarClientFactoryTests.java | 63 ++++++++++++++++++ .../pulsar/listener/PulsarListenerTests.java | 7 +- .../ObservationIntegrationTests.java | 6 +- .../pulsar/observation/ObservationTests.java | 7 +- .../pulsar/reader/PulsarReaderTests.java | 6 +- 11 files changed, 213 insertions(+), 84 deletions(-) delete mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientBuilderCustomizer.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientFactory.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index 1199e113..aac37b53 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/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 diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java index 8b11ea08..f2f4829c 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java +++ b/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 diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java deleted file mode 100644 index 823b1cff..00000000 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarClientFactoryBean.java +++ /dev/null @@ -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 { - - private final LogAccessor logger = new LogAccessor(this.getClass()); - - private final Map config = new HashMap<>(); - - public PulsarClientFactoryBean(Map 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(); - } - } - -} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java new file mode 100644 index 00000000..b6eab137 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientBuilderCustomizer.java new file mode 100644 index 00000000..77a6020d --- /dev/null +++ b/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); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientFactory.java new file mode 100644 index 00000000..7d675bed --- /dev/null +++ b/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; + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java new file mode 100644 index 00000000..7d220b4d --- /dev/null +++ b/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?"); + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 01cf3011..106413db 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/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 diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java index 4806e3e5..cae76088 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java +++ b/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 diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java index 6a442389..0182a23a 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java +++ b/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") diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java index a8475500..535ed90f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java +++ b/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