From c043d8c12652094b4be621867c32da2ede1f3a2f Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sun, 7 May 2023 10:34:12 -0500 Subject: [PATCH] Port auto-config changes from 0.2.x (#402) * Port "Use builder to autoconfigure PulsarProducerFactory (#397)" from 0.2.x * Port "Use builder to autoconfigure PulsarConsumerFactory (#399)" from 0.2.x * Port "Use builder to autoconfigure PulsarReaderFactory (#400)" from 0.2.x * Port "Use builder to autoconfigure PulsarAdministration (#401)" from 0.2.x --- .../listener/ReactivePulsarListenerTests.java | 38 +++--- .../core/CachingPulsarProducerFactory.java | 20 ++-- .../core/DefaultPulsarClientFactory.java | 2 +- .../core/DefaultPulsarConsumerFactory.java | 45 ++++--- .../core/DefaultPulsarProducerFactory.java | 88 ++++++++++---- .../core/DefaultPulsarReaderFactory.java | 47 ++++++-- .../core/PulsarAdminBuilderCustomizer.java | 35 ++++++ .../pulsar/core/PulsarAdministration.java | 69 +++-------- .../pulsar/core/PulsarConsumerFactory.java | 6 - .../pulsar/core/PulsarProducerFactory.java | 9 +- .../pulsar/core/PulsarTemplate.java | 2 +- .../CachingPulsarProducerFactoryTests.java | 16 +-- .../core/ConsumerAcknowledgmentTests.java | 95 +++++---------- .../DefaultPulsarConsumerFactoryTests.java | 13 +- .../DefaultPulsarProducerFactoryTests.java | 14 +-- .../core/DefaultPulsarReaderFactoryTests.java | 71 +++++++++-- .../pulsar/core/FailoverConsumerTests.java | 18 +-- .../core/PulsarAdministrationTests.java | 22 +--- .../core/PulsarProducerFactoryTests.java | 28 +++-- .../pulsar/core/PulsarTemplateTests.java | 29 ++--- .../core/SharedSubscriptionConsumerTests.java | 19 +-- ...efaultPulsarConsumerErrorHandlerTests.java | 90 +++++++------- ...ltPulsarMessageListenerContainerTests.java | 96 +++++++-------- .../pulsar/listener/PulsarListenerTests.java | 112 +++++++----------- .../ObservationIntegrationTests.java | 9 +- .../pulsar/observation/ObservationTests.java | 9 +- ...aultPulsarMessageReaderContainerTests.java | 25 ++-- .../pulsar/reader/PulsarReaderTests.java | 15 +-- 28 files changed, 520 insertions(+), 522 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdminBuilderCustomizer.java 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 f2f4829c..1db5ec0f 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 @@ -20,8 +20,6 @@ import static org.assertj.core.api.Assertions.assertThat; import java.nio.charset.StandardCharsets; import java.time.Duration; -import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Objects; import java.util.concurrent.BlockingQueue; @@ -30,7 +28,6 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; -import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; @@ -107,7 +104,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient, new HashMap<>()); + return new DefaultPulsarProducerFactory<>(pulsarClient); } @Bean @@ -139,8 +136,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Bean PulsarAdministration pulsarAdministration() { - return new PulsarAdministration( - PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())); + return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); } @Bean @@ -347,8 +343,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void jsonSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("json-topic", new User("Jason", i), JSONSchema.of(User.class)); @@ -358,8 +353,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void avroSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("avro-topic", new User("Avi", i), AvroSchema.of(User.class)); @@ -370,7 +364,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void keyvalueSchema() throws Exception { PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( - pulsarClient, Collections.emptyMap()); + pulsarClient); PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE); @@ -382,8 +376,8 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void protobufSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("protobuf-topic", Proto.Person.newBuilder().setId(i).setName("Paul").build(), @@ -494,8 +488,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void jsonSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("json-custom-schema-topic", new User2("Jason", i), JSONSchema.of(User2.class)); @@ -505,8 +498,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void avroSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("avro-custom-schema-topic", new User("Avi", i), AvroSchema.of(User.class)); @@ -517,7 +509,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void keyvalueSchema() throws Exception { PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( - pulsarClient, Collections.emptyMap()); + pulsarClient); PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class), KeyValueEncodingType.INLINE); @@ -530,8 +522,8 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void protobufSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("protobuf-custom-schema-topic", @@ -664,8 +656,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void complexMessageTypeTopicMapping() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); Schema schema = Schema.JSON(User2.class); for (int i = 0; i < 3; i++) { @@ -676,8 +667,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Test void primitiveMessageTypeTopicMapping() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("rplt-topicMapping-string-topic", "Susan " + i, Schema.STRING); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java index 9131eb5e..8678ee20 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java @@ -20,7 +20,6 @@ import java.time.Duration; import java.util.Collection; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; @@ -67,16 +66,17 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor * Construct a caching producer factory with the specified values for the cache * configuration. * @param pulsarClient the client used to create the producers - * @param producerConfig the configuration to use when creating a producer + * @param defaultTopic the default topic to use for the producers + * @param defaultConfigCustomizer the default configuration to apply to the producers * @param topicResolver the topic resolver to use * @param cacheExpireAfterAccess time period to expire unused entries in the cache * @param cacheMaximumSize maximum size of cache (entries) * @param cacheInitialCapacity the initial size of cache */ - public CachingPulsarProducerFactory(PulsarClient pulsarClient, Map producerConfig, - TopicResolver topicResolver, Duration cacheExpireAfterAccess, Long cacheMaximumSize, - Integer cacheInitialCapacity) { - super(pulsarClient, producerConfig, topicResolver); + public CachingPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic, + ProducerBuilderCustomizer defaultConfigCustomizer, TopicResolver topicResolver, + Duration cacheExpireAfterAccess, Long cacheMaximumSize, Integer cacheInitialCapacity) { + super(pulsarClient, defaultTopic, defaultConfigCustomizer, topicResolver); var cacheFactory = CacheProviderFactory., Producer>load(); this.producerCache = cacheFactory.create(cacheExpireAfterAccess, cacheMaximumSize, cacheInitialCapacity, (key, producer, cause) -> { @@ -90,8 +90,8 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor protected Producer doCreateProducer(Schema schema, @Nullable String topic, @Nullable Collection encryptionKeys, @Nullable List> customizers) { Objects.requireNonNull(schema, "Schema must be specified"); - String resolveTopicName = resolveTopicName(topic); - ProducerCacheKey producerCacheKey = new ProducerCacheKey<>(schema, resolveTopicName, + var resolveTopicName = resolveTopicName(topic); + var producerCacheKey = new ProducerCacheKey<>(schema, resolveTopicName, encryptionKeys == null ? null : new HashSet<>(encryptionKeys), customizers); return this.producerCache.getOrCreateIfAbsent(producerCacheKey, (st) -> createCacheableProducer(st.schema, st.topic, st.encryptionKeys, customizers)); @@ -100,7 +100,7 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor private Producer createCacheableProducer(Schema schema, String topic, @Nullable Collection encryptionKeys, @Nullable List> customizers) { try { - Producer producer = super.doCreateProducer(schema, topic, encryptionKeys, customizers); + var producer = super.doCreateProducer(schema, topic, encryptionKeys, customizers); return new ProducerWithCloseCallback<>(producer, (p) -> this.logger.trace(() -> "Client closed producer %s but will skip actual closing" .formatted(ProducerUtils.formatProducer(producer)))); @@ -174,7 +174,7 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor if (o == null || getClass() != o.getClass()) { return false; } - ProducerCacheKey that = (ProducerCacheKey) o; + var that = (ProducerCacheKey) o; return this.topic.equals(that.topic) && this.schemaHash.equals(that.schemaHash) && Objects.equals(this.encryptionKeys, that.encryptionKeys) && Objects.equals(this.customizers, that.customizers); 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 index b6eab137..77921700 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java @@ -37,7 +37,7 @@ public class DefaultPulsarClientFactory implements PulsarClientFactory { * @param serviceUrl the service url */ public DefaultPulsarClientFactory(String serviceUrl) { - this((clientBuilder -> clientBuilder.serviceUrl(serviceUrl))); + this((clientBuilder) -> clientBuilder.serviceUrl(serviceUrl)); } /** diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java index d4e7184c..96c3cd66 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.core; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -30,6 +29,7 @@ import org.apache.pulsar.client.api.ConsumerBuilder; 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.impl.ConsumerBuilderImpl; import org.springframework.lang.Nullable; import org.springframework.util.CollectionUtils; @@ -41,22 +41,25 @@ import org.springframework.util.CollectionUtils; * @author Soby Chacko * @author Alexander Preuß * @author Christophe Bornet + * @author Chris Bono */ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory { - private final Map consumerConfig; - private final PulsarClient pulsarClient; + @Nullable + private final ConsumerBuilderCustomizer defaultConfigCustomizer; + /** * Construct a consumer factory instance. * @param pulsarClient the client used to consume - * @param consumerConfig default configuration to apply to the created consumer or - * empty map to use no default configuration + * @param defaultConfigCustomizer the default configuration to apply to the consumers + * or null to use no default configuration */ - public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, Map consumerConfig) { + public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, + ConsumerBuilderCustomizer defaultConfigCustomizer) { this.pulsarClient = pulsarClient; - this.consumerConfig = Collections.unmodifiableMap(consumerConfig); + this.defaultConfigCustomizer = defaultConfigCustomizer; } @Override @@ -72,25 +75,35 @@ public class DefaultPulsarConsumerFactory implements PulsarConsumerFactory @Nullable List> customizers) throws PulsarClientException { Objects.requireNonNull(schema, "Schema must be specified"); ConsumerBuilder consumerBuilder = this.pulsarClient.newConsumer(schema); - Map config = new HashMap<>(this.consumerConfig); - if (topics != null) { - config.put("topicNames", new HashSet<>(topics)); + + // Apply the default config customizer (preserve the topic) + if (this.defaultConfigCustomizer != null) { + this.defaultConfigCustomizer.customize(consumerBuilder); } - if (metadataProperties != null) { - config.put("properties", new TreeMap<>(metadataProperties)); + if (topics != null) { + replaceTopicsOnBuilder(consumerBuilder, topics); } if (subscriptionName != null) { - config.put("subscriptionName", subscriptionName); + consumerBuilder.subscriptionName(subscriptionName); + } + if (metadataProperties != null) { + replaceMetadataPropertiesOnBuilder(consumerBuilder, metadataProperties); } - ConsumerBuilderConfigurationUtil.loadConf(consumerBuilder, config); if (!CollectionUtils.isEmpty(customizers)) { customizers.forEach(customizer -> customizer.customize(consumerBuilder)); } return consumerBuilder.subscribe(); } - public Map getConsumerConfig() { - return this.consumerConfig; + private void replaceTopicsOnBuilder(ConsumerBuilder builder, Collection topics) { + var builderImpl = (ConsumerBuilderImpl) builder; + builderImpl.getConf().setTopicNames(new HashSet<>(topics)); + } + + private void replaceMetadataPropertiesOnBuilder(ConsumerBuilder builder, + Map metadataProperties) { + var builderImpl = (ConsumerBuilderImpl) builder; + builderImpl.getConf().setProperties(new TreeMap<>(metadataProperties)); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index aa1bf07c..26f6d473 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -18,9 +18,8 @@ package org.springframework.pulsar.core; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; +import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Objects; import org.apache.pulsar.client.api.Producer; @@ -28,6 +27,7 @@ import org.apache.pulsar.client.api.ProducerBuilder; 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.impl.ProducerBuilderImpl; import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; @@ -46,20 +46,58 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory private final LogAccessor logger = new LogAccessor(this.getClass()); - private final Map producerConfig; - private final PulsarClient pulsarClient; + @Nullable + private final String defaultTopic; + + @Nullable + private final ProducerBuilderCustomizer defaultConfigCustomizer; + private final TopicResolver topicResolver; - public DefaultPulsarProducerFactory(PulsarClient pulsarClient, Map config) { - this(pulsarClient, config, new DefaultTopicResolver()); + /** + * Construct a producer factory that uses a default topic resolver. + * @param pulsarClient the client used to create the producers + */ + public DefaultPulsarProducerFactory(PulsarClient pulsarClient) { + this(pulsarClient, null, (pb) -> { + }, new DefaultTopicResolver()); } - public DefaultPulsarProducerFactory(PulsarClient pulsarClient, Map config, - TopicResolver topicResolver) { + /** + * Construct a producer factory that uses a default topic resolver. + * @param pulsarClient the client used to create the producers + * @param defaultTopic the default topic to use for the producers + */ + public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic) { + this(pulsarClient, defaultTopic, (pb) -> { + }, new DefaultTopicResolver()); + } + + /** + * Construct a producer factory that uses a default topic resolver. + * @param pulsarClient the client used to create the producers + * @param defaultTopic the default topic to use for the producers + * @param defaultConfigCustomizer the default configuration to apply to the producers + */ + public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic, + @Nullable ProducerBuilderCustomizer defaultConfigCustomizer) { + this(pulsarClient, defaultTopic, defaultConfigCustomizer, new DefaultTopicResolver()); + } + + /** + * Construct a producer factory that uses the specified parameters. + * @param pulsarClient the client used to create the producers + * @param defaultTopic the default topic to use for the producers + * @param defaultConfigCustomizer the default configuration to apply to the producers + * @param topicResolver the topic resolver to use + */ + public DefaultPulsarProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic, + @Nullable ProducerBuilderCustomizer defaultConfigCustomizer, TopicResolver topicResolver) { this.pulsarClient = pulsarClient; - this.producerConfig = Collections.unmodifiableMap(config); + this.defaultTopic = defaultTopic; + this.defaultConfigCustomizer = defaultConfigCustomizer; this.topicResolver = topicResolver; } @@ -99,36 +137,42 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory @Nullable Collection encryptionKeys, @Nullable List> customizers) throws PulsarClientException { Objects.requireNonNull(schema, "Schema must be specified"); - String resolvedTopic = resolveTopicName(topic); + var resolvedTopic = resolveTopicName(topic); this.logger.trace(() -> "Creating producer for '%s' topic".formatted(resolvedTopic)); - ProducerBuilder producerBuilder = this.pulsarClient.newProducer(schema); + var producerBuilder = this.pulsarClient.newProducer(schema); - Map config = new HashMap<>(this.producerConfig); - - // Replace default keys - workaround as they can't be replaced through the builder - if (encryptionKeys != null) { - config.put("encryptionKeys", encryptionKeys); + // Apply the default config customizer (preserve the topic) + if (this.defaultConfigCustomizer != null) { + this.defaultConfigCustomizer.customize(producerBuilder); } - ProducerBuilderConfigurationUtil.loadConf(producerBuilder, config); producerBuilder.topic(resolvedTopic); + // Replace default keys - workaround as they can't be replaced through the builder + maybeSetEncryptionKeys(producerBuilder, encryptionKeys); + + // Apply any user-specified customizers (preserve the topic) if (!CollectionUtils.isEmpty(customizers)) { customizers.forEach((c) -> c.customize(producerBuilder)); } - // make sure the customizer do not override the topic producerBuilder.topic(resolvedTopic); return producerBuilder.create(); } protected String resolveTopicName(String userSpecifiedTopic) { - String defaultTopic = Objects.toString(getProducerConfig().get("topicName"), null); - return this.topicResolver.resolveTopic(userSpecifiedTopic, () -> defaultTopic).orElseThrow(); + return this.topicResolver.resolveTopic(userSpecifiedTopic, this::getDefaultTopic).orElseThrow(); } @Override - public Map getProducerConfig() { - return this.producerConfig; + public String getDefaultTopic() { + return this.defaultTopic; + } + + private void maybeSetEncryptionKeys(ProducerBuilder builder, @Nullable Collection encryptionKeys) { + if (encryptionKeys != null) { + var builderImpl = (ProducerBuilderImpl) builder; + builderImpl.getConf().setEncryptionKeys(new HashSet<>(encryptionKeys)); + } } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java index fe3f076e..f07d450a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java @@ -16,9 +16,9 @@ package org.springframework.pulsar.core; -import java.util.Collections; +import java.util.Collection; +import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Objects; import org.apache.pulsar.client.api.MessageId; @@ -27,6 +27,7 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.ReaderBuilder; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.ReaderBuilderImpl; import org.springframework.lang.Nullable; import org.springframework.util.CollectionUtils; @@ -41,15 +42,27 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { private final PulsarClient pulsarClient; - private final Map readerConfig; + @Nullable + private final ReaderBuilderCustomizer defaultConfigCustomizer; + /** + * Construct a reader factory instance with no default configuration. + * @param pulsarClient the client used to consume + */ public DefaultPulsarReaderFactory(PulsarClient pulsarClient) { - this(pulsarClient, Collections.emptyMap()); + this(pulsarClient, null); } - public DefaultPulsarReaderFactory(PulsarClient pulsarClient, Map readerConfig) { + /** + * Construct a reader factory instance. + * @param pulsarClient the client used to consume + * @param defaultConfigCustomizer the default configuration to apply to the readers or + * null to use no default configuration + */ + public DefaultPulsarReaderFactory(PulsarClient pulsarClient, + @Nullable ReaderBuilderCustomizer defaultConfigCustomizer) { this.pulsarClient = pulsarClient; - this.readerConfig = readerConfig; + this.defaultConfigCustomizer = defaultConfigCustomizer; } @Override @@ -57,12 +70,19 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { @Nullable List> customizers) throws PulsarClientException { Objects.requireNonNull(schema, "Schema must be specified"); ReaderBuilder readerBuilder = this.pulsarClient.newReader(schema); - if (!CollectionUtils.isEmpty(topics)) { - readerBuilder.topics(topics); - } - readerBuilder.startMessageId(messageId); - readerBuilder.loadConf(this.readerConfig); + // Apply the default config customizer (preserve the topics) + if (this.defaultConfigCustomizer != null) { + this.defaultConfigCustomizer.customize(readerBuilder); + } + + if (!CollectionUtils.isEmpty(topics)) { + replaceTopicsOnBuilder(readerBuilder, topics); + } + + if (messageId != null) { + readerBuilder.startMessageId(messageId); + } if (!CollectionUtils.isEmpty(customizers)) { customizers.forEach(customizer -> customizer.customize(readerBuilder)); @@ -71,4 +91,9 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { return readerBuilder.create(); } + private void replaceTopicsOnBuilder(ReaderBuilder builder, Collection topics) { + var builderImpl = (ReaderBuilderImpl) builder; + builderImpl.getConf().setTopicNames(new HashSet<>(topics)); + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdminBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdminBuilderCustomizer.java new file mode 100644 index 00000000..6315159e --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdminBuilderCustomizer.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.admin.PulsarAdminBuilder; + +/** + * The interface to customize a {@link PulsarAdminBuilder}. + * + * @author Chris Bono + */ +@FunctionalInterface +public interface PulsarAdminBuilderCustomizer { + + /** + * Customizes a {@link PulsarAdminBuilder}. + * @param adminBuilder the builder to customize + */ + void customize(PulsarAdminBuilder adminBuilder); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java index 777096ce..2f7f65dc 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java @@ -18,20 +18,16 @@ package org.springframework.pulsar.core; import java.util.Arrays; import java.util.Collection; -import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; -import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import org.apache.pulsar.client.admin.PulsarAdmin; -import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException; import org.springframework.beans.BeansException; import org.springframework.beans.factory.SmartInitializingSingleton; @@ -40,7 +36,6 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; import org.springframework.util.CollectionUtils; -import org.springframework.util.StringUtils; /** * An administration class that delegates to {@link PulsarAdmin} to create and manage @@ -55,28 +50,27 @@ public class PulsarAdministration private final LogAccessor logger = new LogAccessor(this.getClass()); - private final PulsarAdminBuilder adminBuilder; - @Nullable private ApplicationContext applicationContext; + @Nullable + private final PulsarAdminBuilderCustomizer adminCustomizer; + /** - * Construct a {@code PulsarAdministration} instance using the given configuration for - * the underlying {@link PulsarAdmin}. - * @param adminConfig the {@link PulsarAdmin} configuration + * Construct a default instance using the specified service url. + * @param serviceHttpUrl the admin http service url */ - public PulsarAdministration(Map adminConfig) { - this.adminBuilder = PulsarAdmin.builder(); - loadConf(this.adminBuilder, adminConfig); + public PulsarAdministration(String serviceHttpUrl) { + this((adminBuilder) -> adminBuilder.serviceHttpUrl(serviceHttpUrl)); } /** - * Construct a {@code PulsarAdministration} instance using the given builder for the - * underlying {@link PulsarAdmin}. - * @param adminBuilder the {@link PulsarAdminBuilder} + * Construct an instance with the specified customizations. + * @param adminCustomizer the customizer to apply to the builder or null to use the + * default admin builder without modifications */ - public PulsarAdministration(PulsarAdminBuilder adminBuilder) { - this.adminBuilder = adminBuilder; + public PulsarAdministration(@Nullable PulsarAdminBuilderCustomizer adminCustomizer) { + this.adminCustomizer = adminCustomizer; } @Override @@ -89,39 +83,6 @@ public class PulsarAdministration this.applicationContext = applicationContext; } - private void loadConf(PulsarAdminBuilder builder, Map adminConfig) { - var conf = new HashMap<>(adminConfig); - - // Workaround the fact that the PulsarAdminImpl does not attempt to construct the - // timeout settings from the config props - if (conf.remove("connectionTimeoutMs") instanceof Integer connectTimeout) { - builder.connectionTimeout(connectTimeout, TimeUnit.MILLISECONDS); - } - if (conf.remove("readTimeoutMs") instanceof Integer readTimeout) { - builder.readTimeout(readTimeout, TimeUnit.MILLISECONDS); - } - if (conf.remove("requestTimeoutMs") instanceof Integer requestTimeout) { - builder.requestTimeout(requestTimeout, TimeUnit.MILLISECONDS); - } - if (conf.remove("autoCertRefreshSeconds") instanceof Integer autoCertRefreshTime) { - builder.autoCertRefreshTime(autoCertRefreshTime, TimeUnit.SECONDS); - } - builder.loadConf(conf); - - // Workaround the fact that the PulsarAdminImpl does not attempt to construct the - // authentication from the config props - var authPluginClassName = (String) conf.get("authPluginClassName"); - var authParams = (String) conf.get("authParams"); - if (StringUtils.hasText(authPluginClassName) && StringUtils.hasText(authParams)) { - try { - builder.authentication(authPluginClassName, authParams); - } - catch (UnsupportedAuthenticationException ex) { - throw new RuntimeException("Unable to create admin auth: " + ex.getMessage(), ex); - } - } - } - private void initialize() { var topics = Objects.requireNonNull(this.applicationContext, "Application context was not set") .getBeansOfType(PulsarTopic.class, false, false).values(); @@ -129,7 +90,11 @@ public class PulsarAdministration } public PulsarAdmin createAdminClient() throws PulsarClientException { - return this.adminBuilder.build(); + var adminBuilder = PulsarAdmin.builder(); + if (this.adminCustomizer != null) { + this.adminCustomizer.customize(adminBuilder); + } + return adminBuilder.build(); } @Override diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java index c3bc1c08..ec2b529f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarConsumerFactory.java @@ -85,10 +85,4 @@ public interface PulsarConsumerFactory { @Nullable Map metadataProperties, @Nullable List> customizers) throws PulsarClientException; - /** - * Return the configuration options to use when creating consumers. - * @return the configuration options - */ - Map getConsumerConfig(); - } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index 984c74d0..3fb6a1ef 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.core; import java.util.Collection; import java.util.List; -import java.util.Map; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerBuilder; @@ -78,9 +77,11 @@ public interface PulsarProducerFactory { @Nullable List> customizers) throws PulsarClientException; /** - * Return a map of configuration options to use when creating producers. - * @return the map of configuration options + * Get the default topic to use for all created producers. + * @return the default topic to use for all created producers or null if no default + * set */ - Map getProducerConfig(); + @Nullable + String getDefaultTopic(); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java index e005a7e7..bd023f30 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java @@ -221,7 +221,7 @@ public class PulsarTemplate @Nullable Schema schema, @Nullable Collection encryptionKeys, @Nullable TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, @Nullable ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { - String defaultTopic = Objects.toString(this.producerFactory.getProducerConfig().get("topicName"), null); + String defaultTopic = Objects.toString(this.producerFactory.getDefaultTopic(), null); String topicName = this.topicResolver.resolveTopic(topic, message, () -> defaultTopic).orElseThrow(); this.logger.trace(() -> "Sending msg to '%s' topic".formatted(topicName)); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java index 5c443c93..62b8283f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java @@ -27,7 +27,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.stream.Stream; @@ -46,6 +45,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.springframework.lang.Nullable; import org.springframework.pulsar.core.CachingPulsarProducerFactory.ProducerCacheKey; import org.springframework.pulsar.core.CachingPulsarProducerFactory.ProducerWithCloseCallback; import org.springframework.test.util.ReflectionTestUtils; @@ -167,7 +167,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void factoryDestroyCleansUpCacheAndClosesProducers() throws PulsarClientException { - CachingPulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + CachingPulsarProducerFactory producerFactory = producerFactory(pulsarClient, null, null); var actualProducer1 = actualProducer(producerFactory.createProducer(schema, "topic1")); var actualProducer2 = actualProducer(producerFactory.createProducer(schema, "topic2")); var cacheKey1 = new ProducerCacheKey<>(schema, "topic1", null, null); @@ -183,8 +183,8 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void producerEvictedFromCache() throws PulsarClientException { - CachingPulsarProducerFactory producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap(), new DefaultTopicResolver(), Duration.ofSeconds(3L), 10L, 2); + CachingPulsarProducerFactory producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, null, + null, new DefaultTopicResolver(), Duration.ofSeconds(3L), 10L, 2); var actualProducer = actualProducer(producerFactory.createProducer(schema, "topic1")); var cacheKey = new ProducerCacheKey<>(schema, "topic1", null, null); var producerCache = getAssertedProducerCache(producerFactory, Collections.singletonList(cacheKey)); @@ -198,7 +198,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { void createProducerEncountersException() { pulsarClient = spy(pulsarClient); when(this.pulsarClient.newProducer(schema)).thenThrow(new RuntimeException("5150")); - var producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + var producerFactory = producerFactory(pulsarClient, null, null); assertThatThrownBy(() -> producerFactory.createProducer(schema, "topic1")).isInstanceOf(RuntimeException.class) .hasMessage("5150"); getAssertedProducerCache(producerFactory, Collections.emptyList()); @@ -228,9 +228,9 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Override protected CachingPulsarProducerFactory producerFactory(PulsarClient pulsarClient, - Map producerConfig) { - var producerFactory = new CachingPulsarProducerFactory(pulsarClient, producerConfig, - new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2); + @Nullable String defaultTopic, @Nullable ProducerBuilderCustomizer defaultConfigCustomizer) { + var producerFactory = new CachingPulsarProducerFactory(pulsarClient, defaultTopic, + defaultConfigCustomizer, new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2); producerFactories.add(producerFactory); return producerFactory; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java index f5b5c986..1230ea7a 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java @@ -30,9 +30,7 @@ import static org.mockito.Mockito.verify; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -61,14 +59,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void testRecordAck() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-011"), - "subscriptionName", "cons-ack-tests-sb-011"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-011", "cons-ack-tests-sb-011"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { }); @@ -85,9 +79,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { return invocation.callRealMethod(); }).when(containerConsumer).acknowledge(any(MessageId.class)); - Map prodConfig = Map.of("topicName", "cons-ack-tests-011"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "cons-ack-tests-011"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync("hello john doe"); @@ -99,13 +92,10 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void testBatchAck() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-012"), - "subscriptionName", "cons-ack-tests-sb-012"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-012", "cons-ack-tests-sb-012"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); CountDownLatch latch = new CountDownLatch(10); pulsarContainerProperties @@ -115,9 +105,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { pulsarConsumerFactory, pulsarContainerProperties); Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); - Map prodConfig = Map.of("topicName", "cons-ack-tests-012"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "cons-ack-tests-012"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync("hello john doe"); @@ -133,16 +122,12 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void testBatchAckButSomeRecordsFail() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-013"), - "subscriptionName", "cons-ack-tests-sb-013"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-013", "cons-ack-tests-sb-013"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); CountDownLatch latch = new CountDownLatch(10); - pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { latch.countDown(); if (latch.getCount() % 2 == 0) { @@ -160,9 +145,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { return invocation.callRealMethod(); }).when(containerConsumer).acknowledge(any(MessageId.class)); - Map prodConfig = Map.of("topicName", "cons-ack-tests-013"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "cons-ack-tests-013"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync("hello john doe"); @@ -197,22 +181,17 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test @SuppressWarnings("unchecked") void testManualAckForRecordListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-014"), - "subscriptionName", "cons-ack-tests-sb-014"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-014", "cons-ack-tests-sb-014"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); List acksObjects = new ArrayList<>(); PulsarAcknowledgingMessageListener pulsarAcknowledgingMessageListener = (consumer, msg, acknowledgement) -> { acksObjects.add(acknowledgement); acknowledgement.acknowledge(); }; - pulsarContainerProperties.setMessageListener(pulsarAcknowledgingMessageListener); - pulsarContainerProperties.setSchema(Schema.STRING); pulsarContainerProperties.setAckMode(AckMode.MANUAL); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( @@ -226,9 +205,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { return invocation.callRealMethod(); }).when(containerConsumer).acknowledge(any(MessageId.class)); - Map prodConfig = Map.of("topicName", "cons-ack-tests-014"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "cons-ack-tests-014"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync("hello john doe"); @@ -247,34 +225,28 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test @SuppressWarnings("unchecked") void testBatchAckForBatchListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-015"), - "subscriptionName", "cons-ack-tests-sb-015"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-015", "cons-ack-tests-sb-015"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); CountDownLatch latch = new CountDownLatch(1); PulsarBatchMessageListener pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class); - doAnswer(invocation -> { latch.countDown(); return null; }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class)); - pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); - Map prodConfig = Map.of("topicName", "cons-ack-tests-015"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "cons-ack-tests-015"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync("hello john doe"); @@ -291,34 +263,28 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test @SuppressWarnings("unchecked") void testBatchNackForEntireBatchWhenUsingBatchListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("cons-ack-tests-016"), - "subscriptionName", "cons-ack-tests-sb-016"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); - DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); - + DefaultPulsarConsumerFactory pulsarConsumerFactory = spy(new DefaultPulsarConsumerFactory<>( + pulsarClient, defaultConfig("cons-ack-tests-016", "cons-ack-tests-sb-016"))); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); pulsarContainerProperties.setBatchTimeoutMillis(60_000); pulsarContainerProperties.setBatchListener(true); PulsarBatchMessageListener pulsarBatchMessageListener = mock(PulsarBatchMessageListener.class); CountDownLatch latch = new CountDownLatch(1); - doAnswer(invocation -> { latch.countDown(); throw new RuntimeException(); }).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class)); - pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); - Map prodConfig = Map.of("topicName", "cons-ack-tests-016"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "cons-ack-tests-016"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync("hello john doe"); @@ -335,26 +301,21 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { @Test void messagesAreProperlyAckdOnContainerStopBeforeExitingListenerThread() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("duplicate-message-test"), - "subscriptionName", "duplicate-sub-1"); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); - + defaultConfig("duplicate-message-test", "duplicate-sub-1")); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); AtomicInteger counter1 = new AtomicInteger(0); - pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - counter1.getAndIncrement(); - }); + pulsarContainerProperties + .setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> counter1.getAndIncrement()); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container1 = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); container1.start(); - Map prodConfig = Collections.singletonMap("topicName", "duplicate-message-test"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "duplicate-message-test"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.send("hello john doe"); @@ -368,9 +329,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { container1.stop(); AtomicInteger counter2 = new AtomicInteger(0); - pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { - counter2.getAndIncrement(); - }); + pulsarContainerProperties + .setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> counter2.getAndIncrement()); pulsarContainerProperties.setSchema(Schema.STRING); DefaultPulsarMessageListenerContainer container2 = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); @@ -389,4 +349,11 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { pulsarClient.close(); } + private ConsumerBuilderCustomizer defaultConfig(String topicName, String subscriptionName) { + return (consumerBuilder) -> { + consumerBuilder.topic(topicName); + consumerBuilder.subscriptionName(subscriptionName); + }; + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java index 83016549..23e5795e 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactoryTests.java @@ -23,7 +23,6 @@ import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import java.util.Collections; -import java.util.HashMap; import java.util.List; import java.util.Map; @@ -72,7 +71,7 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { @BeforeEach void createConsumerFactory() { - consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap()); + consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Test @@ -170,11 +169,11 @@ class DefaultPulsarConsumerFactoryTests implements PulsarTestContainerSupport { @BeforeEach void createConsumerFactory() { - Map defaultConfig = new HashMap<>(); - defaultConfig.put("topicNames", Collections.singleton(defaultTopic)); - defaultConfig.put("properties", defaultMetadataProperties); - defaultConfig.put("subscriptionName", defaultSubscription); - consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, defaultConfig); + consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> { + consumerBuilder.topic(defaultTopic); + consumerBuilder.subscriptionName(defaultSubscription); + consumerBuilder.properties(defaultMetadataProperties); + }); } @Test diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java index 89cebbac..71b157fe 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarProducerFactoryTests.java @@ -18,14 +18,13 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; -import java.util.Map; - import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.junit.jupiter.api.Test; +import org.springframework.lang.Nullable; + /** * Tests for {@link DefaultPulsarProducerFactory}. * @@ -35,8 +34,7 @@ class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void createProducerMultipleTimeDoesNotCacheProducer() throws PulsarClientException { - Map producerConfig = Collections.emptyMap(); - PulsarProducerFactory producerFactory = producerFactory(pulsarClient, producerConfig); + PulsarProducerFactory producerFactory = newProducerFactory(); try (Producer producer1 = producerFactory.createProducer(schema, "topic1")) { try (Producer producer2 = producerFactory.createProducer(schema, "topic1")) { try (Producer producer3 = producerFactory.createProducer(schema, "topic1")) { @@ -47,9 +45,9 @@ class DefaultPulsarProducerFactoryTests extends PulsarProducerFactoryTests { } @Override - protected PulsarProducerFactory producerFactory(PulsarClient pulsarClient, - Map producerConfig) { - return new DefaultPulsarProducerFactory<>(pulsarClient, producerConfig); + protected PulsarProducerFactory producerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic, + @Nullable ProducerBuilderCustomizer defaultConfigCustomizer) { + return new DefaultPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizer); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java index eb59d8da..09f41211 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java @@ -21,7 +21,6 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.api.Message; @@ -65,7 +64,7 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo @BeforeEach void createReaderFactory() { - pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, Collections.emptyMap()); + pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient); } @Test @@ -74,9 +73,8 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo try (Reader reader = pulsarReaderFactory.createReader(List.of("basic-pulsar-reader-topic"), MessageId.earliest, Schema.STRING, Collections.emptyList())) { - Map prodConfig = Map.of("topicName", "basic-pulsar-reader-topic"); PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "basic-pulsar-reader-topic"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.send("hello john doe"); @@ -87,9 +85,8 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo @Test void readingFromTheMiddleOfTheTopic() throws Exception { - Map prodConfig = Map.of("topicName", "reading-from-the-middle-of-topic"); PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "reading-from-the-middle-of-topic"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); MessageId[] messageIds = new MessageId[10]; @@ -113,9 +110,8 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo void readingFromTheEndOfTheTopic() throws Exception { Message message; - Map prodConfig = Map.of("topicName", "basic-pulsar-reader-topic"); PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "basic-pulsar-reader-topic"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.send("hello john doe"); @@ -131,6 +127,63 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo } } + @Test + void useFactoryDefaults() throws Exception { + pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, (readerBuilder) -> { + readerBuilder.topic("basic-pulsar-reader-topic"); + readerBuilder.startMessageId(MessageId.earliest); + }); + // The following code expects the above topic and startMessageId to be used + Message message; + try (Reader reader = pulsarReaderFactory.createReader(null, null, Schema.STRING, + Collections.emptyList())) { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + "basic-pulsar-reader-topic"); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + pulsarTemplate.send("hello john doe"); + message = reader.readNext(); + } + assertThat(message.getValue()).isEqualTo("hello john doe"); + } + + @Test + void overrideFactoryDefaults() throws Exception { + pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, (readerBuilder) -> { + readerBuilder.topic("foo-topic"); + readerBuilder.startMessageId(MessageId.latest); + }); + // The following code expects the above topic and startMessageId to be ignored + // (overridden) + Message message; + try (Reader reader = pulsarReaderFactory.createReader(List.of("basic-pulsar-reader-topic"), + MessageId.earliest, Schema.STRING, Collections.emptyList())) { + + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + "basic-pulsar-reader-topic"); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + pulsarTemplate.send("hello john doe"); + + message = reader.readNext(); + } + assertThat(message.getValue()).isEqualTo("hello john doe"); + } + + @Test + void customizersAreAppliedLast() throws Exception { + ReaderBuilderCustomizer customizer = (readerBuilder) -> readerBuilder + .topic("basic-pulsar-reader-topic"); + // The following code expects the above topic will override the passed in + // 'foo-topic' + try (var reader = pulsarReaderFactory.createReader(List.of("foo-topic"), MessageId.earliest, Schema.STRING, + List.of(customizer))) { + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient, + "basic-pulsar-reader-topic"); + var pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); + pulsarTemplate.send("hello john doe"); + assertThat(reader.readNext().getValue()).isEqualTo("hello john doe"); + } + } + } @Nested @@ -140,7 +193,7 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo @BeforeEach void createReaderFactory() { - pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, Collections.emptyMap()); + pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient); } @Test diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java index 5c720963..8b2f30fd 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java @@ -19,8 +19,6 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; import java.io.Serial; -import java.util.Collections; -import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -48,18 +46,16 @@ class FailoverConsumerTests implements PulsarTestContainerSupport { void testFailOverConsumersOnPartitionedTopic() throws Exception { PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()) .build(); - String topicName = "persistent://public/default/my-part-topic-1"; - int numPartitions = 3; - admin.topics().createPartitionedTopic(topicName, numPartitions); - - Map config = Map.of("topicNames", Collections.singleton("my-part-topic-1"), "subscriptionName", - "my-part-subscription-1"); + admin.topics().createPartitionedTopic(topicName, 3); PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("my-part-topic-1"); + consumerBuilder.subscriptionName("my-part-subscription-1"); + }); CountDownLatch latch1 = new CountDownLatch(1); CountDownLatch latch2 = new CountDownLatch(1); @@ -86,10 +82,8 @@ class FailoverConsumerTests implements PulsarTestContainerSupport { pulsarConsumerFactory, pulsarContainerProperties); container3.start(); - Map prodConfig = Map.of("topicName", "my-part-topic-1", "messageRoutingMode", - MessageRoutingMode.CustomPartition); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "my-part-topic-1", (pb) -> pb.messageRoutingMode(MessageRoutingMode.CustomPartition)); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.newMessage("hello john doe") diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java index 0126f5bf..1598fd33 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java @@ -18,19 +18,13 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; -import static org.assertj.core.api.InstanceOfAssertFactories.type; import java.util.Collections; -import java.util.HashMap; import java.util.List; -import java.util.Map; import org.apache.pulsar.client.admin.PulsarAdmin; -import org.apache.pulsar.client.admin.PulsarAdminBuilder; import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.impl.auth.AuthenticationBasic; -import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -63,19 +57,6 @@ public class PulsarAdministrationTests implements PulsarTestContainerSupport { @Autowired private PulsarAdministration pulsarAdministration; - @Test - void constructorRespectsAuthenticationProps() { - Map props = new HashMap<>(); - props.put("authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationBasic"); - props.put("authParams", "{\"userId\":\"foo\", \"password\":\"bar\"}"); - PulsarAdministration admin = new PulsarAdministration(props); - - assertThat(admin).extracting("adminBuilder").asInstanceOf(type(PulsarAdminBuilder.class)).extracting("conf") - .asInstanceOf(type(ClientConfigurationData.class)) - .extracting(ClientConfigurationData::getAuthentication).isInstanceOf(AuthenticationBasic.class) - .hasFieldOrPropertyWithValue("userId", "foo").hasFieldOrPropertyWithValue("password", "bar"); - } - private void assertThatTopicsExist(List expected) throws PulsarAdminException { assertThatTopicsExistIn(expected, NAMESPACE); } @@ -106,8 +87,7 @@ public class PulsarAdministrationTests implements PulsarTestContainerSupport { @Bean PulsarAdministration pulsarAdministration() { - return new PulsarAdministration( - PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())); + return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java index 884c38ba..8aac0247 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java @@ -26,7 +26,6 @@ import static org.mockito.Mockito.verify; import java.util.Arrays; import java.util.Collections; -import java.util.Map; import java.util.Set; import org.apache.pulsar.client.api.Producer; @@ -42,6 +41,7 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.mockito.InOrder; +import org.springframework.lang.Nullable; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; /** @@ -91,15 +91,15 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport } protected PulsarProducerFactory newProducerFactory() { - return producerFactory(pulsarClient, Collections.emptyMap()); + return producerFactory(pulsarClient, null, null); } protected PulsarProducerFactory newProducerFactoryWithDefaultTopic(String defaultTopic) { - return producerFactory(pulsarClient, Collections.singletonMap("topicName", defaultTopic)); + return producerFactory(pulsarClient, defaultTopic, null); } private PulsarProducerFactory newProducerFactoryWithDefaultKeys(Set defaultKeys) { - return producerFactory(pulsarClient, Collections.singletonMap("encryptionKeys", defaultKeys)); + return producerFactory(pulsarClient, null, (pb) -> defaultKeys.forEach(pb::addEncryptionKey)); } /** @@ -114,11 +114,12 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport /** * Subclasses override to provide concrete {@link PulsarProducerFactory} instance. * @param pulsarClient the Pulsar client - * @param producerConfig the Pulsar producers config + * @param defaultTopic the default topic to use for the producers + * @param defaultConfigCustomizer the default configuration to apply to the producers * @return a Pulsar producer factory instance to use for the tests */ protected abstract PulsarProducerFactory producerFactory(PulsarClient pulsarClient, - Map producerConfig); + @Nullable String defaultTopic, @Nullable ProducerBuilderCustomizer defaultConfigCustomizer); @Test @SuppressWarnings("unchecked") @@ -223,19 +224,20 @@ abstract class PulsarProducerFactoryTests implements PulsarTestContainerSupport @Test void withDefaultEncryptionKeys() throws PulsarClientException { - var keys = Set.of("key"); - var producerFactory = newProducerFactoryWithDefaultKeys(keys); + var defaultKeys = Set.of("default-key"); + var producerFactory = newProducerFactoryWithDefaultKeys(defaultKeys); try (var producer = producerFactory.createProducer(schema, "topic0")) { - assertThatProducerHasEncryptionKeys(producer, keys); + assertThatProducerHasEncryptionKeys(producer, defaultKeys); } } @Test void specificEncryptionKeys() throws PulsarClientException { - var keys = Set.of("key"); - var producerFactory = newProducerFactory(); - try (var producer = producerFactory.createProducer(schema, "topic0", keys, null)) { - assertThatProducerHasEncryptionKeys(producer, keys); + var defaultKeys = Set.of("default-key"); + var userSpecifiedKeys = Set.of("user-key"); + var producerFactory = newProducerFactoryWithDefaultKeys(defaultKeys); + try (var producer = producerFactory.createProducer(schema, "topic0", userSpecifiedKeys, null)) { + assertThatProducerHasEncryptionKeys(producer, userSpecifiedKeys); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index bb4fb27d..217bbc84 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -29,9 +29,7 @@ import static org.mockito.Mockito.when; import java.time.Duration; import java.util.ArrayList; import java.util.Collections; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.UUID; @@ -192,8 +190,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { @ParameterizedTest(name = "{0}") @MethodSource("interceptorInvocationTestProvider") void interceptorInvocationTest(String topic, List interceptors) throws Exception { - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - Collections.singletonMap("topicName", topic)); + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, topic); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors); pulsarTemplate.send("test-interceptor"); for (ProducerInterceptor interceptor : interceptors) { @@ -214,8 +211,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { void sendMessageWithTopicInferredByTypeMappings(boolean producerFactoryHasDefaultTopic) throws Exception { String topic = "ptt-topicInferred-" + producerFactoryHasDefaultTopic + "-topic"; PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - producerFactoryHasDefaultTopic ? Collections.singletonMap("topicName", "fake-topic") - : Collections.emptyMap()); + producerFactoryHasDefaultTopic ? "fake-topic" : null); // Topic mappings allows not specifying the topic when sending (nor having // default on producer) DefaultTopicResolver topicResolver = new DefaultTopicResolver(); @@ -229,8 +225,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { @Test void sendMessageWithoutTopicFails() { - PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, - Collections.emptyMap()); + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send("test-message")) .withMessage("Topic must be specified when no default topic is configured"); @@ -238,11 +233,8 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { private Message sendAndConsume(ThrowingConsumer> sendFunction, String topic, Schema schema, T expectedValue, Boolean withDefaultTopic) throws Exception { - Map config = new HashMap<>(); - if (withDefaultTopic) { - config.put("topicName", topic); - } - PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, config); + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, + withDefaultTopic ? topic : null); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); return sendAndConsume(pulsarTemplate, sendFunction, topic, schema, expectedValue); } @@ -282,8 +274,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { @Test void withSchemaInferredByTypeMappings() throws Exception { String topic = "ptt-schemaInferred-topic"; - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - Collections.singletonMap("topicName", topic)); + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, topic); // Custom schema resolver allows not specifying the schema when sending DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); @@ -301,9 +292,8 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { @Test void sendNullWithDefaultTopicFails() { - HashMap config = new HashMap<>(); - config.put("topicName", "sendNullWithDefaultTopicFails"); - PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, config); + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, + "sendNullWithDefaultTopicFails"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send(null, Schema.STRING)) .withMessage("Topic must be specified when the message is null"); @@ -311,8 +301,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { @Test void sendNullWithoutSchemaFails() { - PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, - Collections.emptyMap()); + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); assertThatIllegalArgumentException() .isThrownBy(() -> pulsarTemplate.send("sendNullWithoutSchemaFails", null, null)) diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java index 455c2048..92212f67 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/SharedSubscriptionConsumerTests.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -57,9 +56,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo try { pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>( - pulsarClient, - Map.of("topicNames", Collections.singleton("shared-subscription-single-msg-test-topic"), - "subscriptionName", "shared-subscription-single-msg-test-sub")); + pulsarClient, (consumerBuilder) -> { + consumerBuilder.topic("shared-subscription-single-msg-test-topic"); + consumerBuilder.subscriptionName("shared-subscription-single-msg-test-sub"); + }); CountDownLatch latch1 = new CountDownLatch(1); CountDownLatch latch2 = new CountDownLatch(1); @@ -76,9 +76,8 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo container3 = createAndStartContainer(pulsarConsumerFactory, latch3, "three", messageCountByKey3, SubscriptionType.Shared); - Map prodConfig = Map.of("topicName", "shared-subscription-single-msg-test-topic"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( - pulsarClient, prodConfig); + pulsarClient, "shared-subscription-single-msg-test-topic"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.newMessage("hello john doe").sendAsync(); @@ -114,8 +113,10 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo try { pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); DefaultPulsarConsumerFactory consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - Map.of("topicNames", Collections.singleton("key-shared-batch-disabled-topic"), "subscriptionName", - "key-shared-batch-disabled-sub")); + (consumerBuilder) -> { + consumerBuilder.topic("key-shared-batch-disabled-topic"); + consumerBuilder.subscriptionName("key-shared-batch-disabled-sub"); + }); CountDownLatch latch = new CountDownLatch(30); @@ -132,7 +133,7 @@ public class SharedSubscriptionConsumerTests implements PulsarTestContainerSuppo Thread.sleep(5_000); DefaultPulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Map.of("topicName", "key-shared-batch-disabled-topic", "batchingEnabled", "false")); + "key-shared-batch-disabled-topic", (pb) -> pb.enableBatching(false)); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.newMessage("alice-" + i) diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java index 47d70346..8fdcac40 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java @@ -28,9 +28,7 @@ import static org.mockito.Mockito.when; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; import org.apache.pulsar.client.api.Consumer; @@ -56,13 +54,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void happyPathErrorHandlingForRecordMessageListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-1"), - "subscriptionName", "default-error-handler-tests-sub-1"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-1"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-1"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); @@ -74,9 +72,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain pulsarContainerProperties.setMessageListener(messageListener); pulsarContainerProperties.setSchema(Schema.STRING); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-1"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-1"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); @@ -105,13 +102,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void errorHandlingForRecordMessageListenerWithTransientError() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-2"), - "subscriptionName", "default-error-handler-tests-sub-2"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-2"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-2"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); @@ -127,9 +124,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain pulsarContainerProperties.setMessageListener(messageListener); pulsarContainerProperties.setSchema(Schema.STRING); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-2"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-2"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class); @@ -152,13 +148,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void everyOtherRecordThrowsNonTransientExceptionsRecordMessageListener() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-3"), - "subscriptionName", "default-error-handler-tests-sub-3"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-3"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-3"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); PulsarRecordMessageListener messageListener = mock(PulsarRecordMessageListener.class); @@ -174,9 +170,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain pulsarContainerProperties.setMessageListener(messageListener); pulsarContainerProperties.setSchema(Schema.INT32); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-3"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-3"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); PulsarTemplate mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS); @@ -209,13 +204,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerFirstOneOnlyErrorAndRecover() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-4"), - "subscriptionName", "default-error-handler-tests-sub-4"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-4"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-4"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -252,9 +247,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain container.start(); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-4"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-4"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync(i); @@ -279,13 +273,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerRecordFailsInTheMiddle() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-5"), - "subscriptionName", "default-error-handler-tests-sub-5"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-5"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-5"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -321,9 +315,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain container.start(); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-5"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-5"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync(i); @@ -347,13 +340,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerRecordFailsTwiceInTheMiddle() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-6"), - "subscriptionName", "default-error-handler-tests-sub-6"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-6"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-6"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -389,9 +382,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain container.start(); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-6"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-6"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync(i); @@ -415,13 +407,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchRecordListenerRecordFailsInTheMiddleButTransientError() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-7"), - "subscriptionName", "default-error-handler-tests-sub-7"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-7"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-7"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -463,9 +455,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain container.start(); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-7"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-7"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync(i); @@ -482,13 +473,13 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain @Test @SuppressWarnings("unchecked") void batchListenerFailsTransientErrorFollowedByNonTransient() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("default-error-handler-tests-8"), - "subscriptionName", "default-error-handler-tests-sub-8"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("default-error-handler-tests-8"); + consumerBuilder.subscriptionName("default-error-handler-tests-sub-8"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMaxNumMessages(10); @@ -533,9 +524,8 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain container.start(); - Map prodConfig = Map.of("topicName", "default-error-handler-tests-8"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "default-error-handler-tests-8"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 10; i++) { pulsarTemplate.sendAsync(i); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java index 3816207d..b0c13e47 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java @@ -26,9 +26,7 @@ import static org.mockito.Mockito.verify; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -65,13 +63,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void basicDefaultConsumer() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("dpmlct-012"), "subscriptionName", - "dpmlct-sb-012"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-012"); + consumerBuilder.subscriptionName("dpmlct-sb-012"); + }); CountDownLatch latch = new CountDownLatch(1); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties @@ -81,9 +79,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS pulsarConsumerFactory, pulsarContainerProperties); container.start(); - Map prodConfig = Map.of("topicName", "dpmlct-012"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "dpmlct-012"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.sendAsync("hello john doe"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); @@ -94,13 +91,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Disabled @Test void containerPauseAndResumeFeatureUsingWaitAndNotify() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("containerPauseResumeWaitNotify-topic"), - "subscriptionName", "containerPauseResumeWaitNotify-sub"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("containerPauseResumeWaitNotify-topic"); + consumerBuilder.subscriptionName("containerPauseResumeWaitNotify-sub"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { }); @@ -163,13 +160,14 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void subscriptionInitialPositionEarliest() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("dpmlct-013"), "subscriptionName", - "dpmlct-sb-013", "subscriptionInitialPosition", SubscriptionInitialPosition.Earliest); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-013"); + consumerBuilder.subscriptionName("dpmlct-sb-013"); + consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + }); CountDownLatch latch = new CountDownLatch(5); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties @@ -178,9 +176,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - Map prodConfig = Map.of("topicName", "dpmlct-013"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "dpmlct-013"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 5; i++) { pulsarTemplate.send("hello john doe" + i); @@ -194,13 +191,13 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void subscriptionInitialPositionDefaultLatest() throws Exception { - Map config = Map.of("topicNames", Collections.singleton("dpmlct-014"), "subscriptionName", - "dpmlct-sb-014"); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-014"); + consumerBuilder.subscriptionName("dpmlct-sb-014"); + }); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); List messages = new ArrayList<>(); pulsarContainerProperties.setMessageListener( @@ -209,9 +206,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( pulsarConsumerFactory, pulsarContainerProperties); - Map prodConfig = Map.of("topicName", "dpmlct-014"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "dpmlct-014"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 5; i++) { pulsarTemplate.send("hello john doe" + i); @@ -228,15 +224,16 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void negativeAckRedeliveryBackoff() throws Exception { - RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000) - .maxDelayMs(5 * 1000).build(); - Map config = Map.of("topicNames", Collections.singleton("dpmlct-015"), "subscriptionName", - "dpmlct-sb-015", "negativeAckRedeliveryBackoff", redeliveryBackoff); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); + RedeliveryBackoff redeliveryBackoff = MultiplierRedeliveryBackoff.builder().minDelayMs(1000) + .maxDelayMs(5 * 1000).build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = spy( - new DefaultPulsarConsumerFactory<>(pulsarClient, config)); + new DefaultPulsarConsumerFactory<>(pulsarClient, (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-015"); + consumerBuilder.subscriptionName("dpmlct-sb-015"); + consumerBuilder.negativeAckRedeliveryBackoff(redeliveryBackoff); + })); CountDownLatch latch = new CountDownLatch(10); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener) (consumer, msg) -> { @@ -252,9 +249,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS Consumer containerConsumer = ConsumerTestUtils.startContainerAndSpyOnConsumer(container); - Map prodConfig = Collections.singletonMap("topicName", "dpmlct-015"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "dpmlct-015"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 5; i++) { pulsarTemplate.send("hello john doe" + i); @@ -275,16 +271,17 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void deadLetterPolicyDefault() throws Exception { - DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1) - .deadLetterTopic("dpmlct-016-dlq-topic").build(); - Map config = Map.of("topicNames", Collections.singleton("dpmlct-016"), "subscriptionName", - "dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy", - deadLetterPolicy); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); + DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(1) + .deadLetterTopic("dpmlct-016-dlq-topic").build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-016"); + consumerBuilder.subscriptionName("dpmlct-sb-016"); + consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS); + consumerBuilder.deadLetterPolicy(deadLetterPolicy); + }); CountDownLatch dlqLatch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(6); @@ -312,9 +309,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS pulsarConsumerFactory, pulsarContainerProperties); container.start(); - Map prodConfig = Collections.singletonMap("topicName", "dpmlct-016"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "dpmlct-016"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 1; i < 6; i++) { pulsarTemplate.send(i); @@ -331,16 +327,17 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS @Test void deadLetterPolicyCustom() throws Exception { - DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic") - .build(); - Map config = Map.of("topicNames", Collections.singleton("dpmlct-017"), "subscriptionName", - "dpmlct-sb-016", "negativeAckRedeliveryDelayMicros", TimeUnit.SECONDS.toMicros(1), "deadLetterPolicy", - deadLetterPolicy); - PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build(); + DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder().maxRedeliverCount(5).deadLetterTopic("dlq-topic") + .build(); DefaultPulsarConsumerFactory pulsarConsumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, - config); + (consumerBuilder) -> { + consumerBuilder.topic("dpmlct-017"); + consumerBuilder.subscriptionName("dpmlct-sb-017"); + consumerBuilder.negativeAckRedeliveryDelay(1L, TimeUnit.SECONDS); + consumerBuilder.deadLetterPolicy(deadLetterPolicy); + }); CountDownLatch dlqLatch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(6); @@ -371,9 +368,8 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS pulsarConsumerFactory, pulsarContainerProperties); container.start(); - Map prodConfig = Collections.singletonMap("topicName", "dpmlct-017"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - prodConfig); + "dpmlct-017"); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 1; i < 6; i++) { pulsarTemplate.send(i); 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 106413db..405e25a2 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 @@ -22,17 +22,13 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.Message; @@ -105,8 +101,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - Map config = Map.of("topicName", "foo-1"); - return new DefaultPulsarProducerFactory<>(pulsarClient, config); + return new DefaultPulsarProducerFactory<>(pulsarClient, "foo-1"); } @Bean @@ -121,22 +116,19 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - Map config = new HashMap<>(); - return new DefaultPulsarConsumerFactory<>(pulsarClient, config); + return new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Bean PulsarListenerContainerFactory pulsarListenerContainerFactory( PulsarConsumerFactory pulsarConsumerFactory) { - ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>( - pulsarConsumerFactory, new PulsarContainerProperties()); - return pulsarListenerContainerFactory; + return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, + new PulsarContainerProperties()); } @Bean PulsarAdministration pulsarAdministration() { - return new PulsarAdministration( - PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())); + return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); } @Bean @@ -184,12 +176,10 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void concurrencyOnPulsarListenerWithFailoverSubscription(@Autowired PulsarListenerEndpointRegistry registry) throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Map.of("batchingEnabled", false)); - PulsarTemplate customTemplate = new PulsarTemplate<>(pulsarProducerFactory); - - ConcurrentPulsarMessageListenerContainer bar = (ConcurrentPulsarMessageListenerContainer) registry - .getListenerContainer("bar"); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient, null, + (pb) -> pb.enableBatching(false)); + var customTemplate = new PulsarTemplate<>(pulsarProducerFactory); + var bar = (ConcurrentPulsarMessageListenerContainer) registry.getListenerContainer("bar"); assertThat(bar.getConcurrency()).isEqualTo(3); @@ -203,12 +193,10 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void nonDefaultConcurrencySettingNotAllowedOnExclusiveSubscriptions( @Autowired PulsarListenerEndpointRegistry registry) throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Map.of("batchingEnabled", false)); - PulsarTemplate customTemplate = new PulsarTemplate<>(pulsarProducerFactory); - - ConcurrentPulsarMessageListenerContainer bar = (ConcurrentPulsarMessageListenerContainer) registry - .getListenerContainer("bar"); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient, null, + (pb) -> pb.enableBatching(false)); + var customTemplate = new PulsarTemplate<>(pulsarProducerFactory); + var bar = (ConcurrentPulsarMessageListenerContainer) registry.getListenerContainer("bar"); assertThat(bar.getConcurrency()).isEqualTo(3); @@ -460,10 +448,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void jsonSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); - Schema schema = JSONSchema.of(User.class); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var schema = JSONSchema.of(User.class); for (int i = 0; i < 3; i++) { template.send("json-topic", new User("Jason", i), schema); } @@ -473,10 +460,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void avroSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); - Schema schema = AvroSchema.of(User.class); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var schema = AvroSchema.of(User.class); for (int i = 0; i < 3; i++) { template.send("avro-topic", new User("Avi", i), schema); } @@ -486,11 +472,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void keyvalueSchema() throws Exception { - PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( - pulsarClient, Collections.emptyMap()); - PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); - Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, - KeyValueEncodingType.INLINE); + var pulsarProducerFactory = new DefaultPulsarProducerFactory>(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE); for (int i = 0; i < 3; i++) { template.send("keyvalue-topic", new KeyValue<>("Kevin", i), kvSchema); } @@ -500,10 +484,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void protobufSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); - Schema schema = ProtobufSchema.of(Proto.Person.class); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var schema = ProtobufSchema.of(Proto.Person.class); for (int i = 0; i < 3; i++) { template.send("protobuf-topic", Proto.Person.newBuilder().setId(i).setName("Paul").build(), schema); } @@ -640,10 +623,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void jsonSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); - Schema schema = Schema.JSON(User2.class); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var schema = Schema.JSON(User2.class); for (int i = 0; i < 3; i++) { template.send("json-custom-mappings-topic", new User2("Jason", i), schema); } @@ -652,10 +634,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void avroSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); - Schema schema = AvroSchema.of(User.class); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var schema = AvroSchema.of(User.class); for (int i = 0; i < 3; i++) { template.send("avro-custom-mappings-topic", new User("Avi", i), schema); } @@ -664,11 +645,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void keyvalueSchema() throws Exception { - PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( - pulsarClient, Collections.emptyMap()); - PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); - Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class), - KeyValueEncodingType.INLINE); + var pulsarProducerFactory = new DefaultPulsarProducerFactory>(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class), KeyValueEncodingType.INLINE); for (int i = 0; i < 3; i++) { template.send("keyvalue-custom-mappings-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150)), kvSchema); @@ -678,10 +657,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void protobufSchema() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); - Schema schema = ProtobufSchema.of(Proto.Person.class); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var schema = ProtobufSchema.of(Proto.Person.class); for (int i = 0; i < 3; i++) { template.send("protobuf-custom-mappings-topic", Proto.Person.newBuilder().setId(i).setName("Paul").build(), schema); @@ -749,10 +727,9 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void complexMessageTypeTopicMapping() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); - Schema schema = Schema.JSON(User2.class); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); + var schema = Schema.JSON(User2.class); for (int i = 0; i < 3; i++) { template.send("plt-topicMapping-user-topic", new User2("Jason", i), schema); } @@ -761,9 +738,8 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Test void primitiveMessageTypeTopicMapping() throws Exception { - PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, - Collections.emptyMap()); - PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + var pulsarProducerFactory = new DefaultPulsarProducerFactory(pulsarClient); + var template = new PulsarTemplate<>(pulsarProducerFactory); for (int i = 0; i < 3; i++) { template.send("plt-topicMapping-string-topic", "Susan " + i, Schema.STRING); } @@ -1106,9 +1082,7 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Bean public ConsumerBuilderCustomizer myCustomizer() { - return cb -> { - cb.subscriptionName("test-changed-subscription-name"); - }; + return cb -> cb.subscriptionName("test-changed-subscription-name"); } } 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 cae76088..5cacd885 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 @@ -18,14 +18,12 @@ package org.springframework.pulsar.observation; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; @@ -131,7 +129,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap()); + return new DefaultPulsarProducerFactory<>(pulsarClient); } @Bean @@ -147,7 +145,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap()); + return new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Bean @@ -160,8 +158,7 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul @Bean PulsarAdministration pulsarAdministration() { - return new PulsarAdministration( - PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())); + return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); } @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 0182a23a..a63aba68 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 @@ -20,13 +20,11 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import java.util.Arrays; -import java.util.Collections; import java.util.Deque; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; @@ -170,7 +168,7 @@ public class ObservationTests implements PulsarTestContainerSupport { @Bean PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap()); + return new DefaultPulsarProducerFactory<>(pulsarClient); } @Bean @@ -197,7 +195,7 @@ public class ObservationTests implements PulsarTestContainerSupport { @Bean PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, Collections.emptyMap()); + return new DefaultPulsarConsumerFactory<>(pulsarClient, null); } @Bean @@ -224,8 +222,7 @@ public class ObservationTests implements PulsarTestContainerSupport { @Bean PulsarAdministration pulsarAdministration() { - return new PulsarAdministration( - PulsarAdmin.builder().serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl())); + return new PulsarAdministration(PulsarTestContainerSupport.getHttpServiceUrl()); } @Bean diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java index 3c878c40..3af04a72 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainerTests.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.reader; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -44,6 +43,7 @@ import org.springframework.pulsar.test.support.PulsarTestContainerSupport; * Basic tests for {@link DefaultPulsarMessageReaderContainer}. * * @author Soby Chacko + * @author Chris Bono */ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestContainerSupport { @@ -67,9 +67,12 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta @Test void basicDefaultReader() throws Exception { var latch = new CountDownLatch(1); - var config = Map.of("topicNames", Collections.singleton("dprlct-001"), "subscriptionName", "dprlct-sub-001"); - DefaultPulsarReaderFactory pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, config); + DefaultPulsarReaderFactory pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, + (readerBuilder -> { + readerBuilder.topic("dprlct-001"); + readerBuilder.subscriptionName("dprlct-sub-001"); + })); var readerContainerProperties = new PulsarReaderContainerProperties(); readerContainerProperties.setReaderListener((ReaderListener) (reader, msg) -> { assertThat(msg.getValue()).isEqualTo("hello john doe"); @@ -83,9 +86,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta container = new DefaultPulsarMessageReaderContainer<>(pulsarReaderFactory, readerContainerProperties); container.start(); - Map prodConfig = Map.of("topicName", "dprlct-001"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( - pulsarClient, prodConfig); + pulsarClient, "dprlct-001", (pb) -> pb.topic("dprlct-001")); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.sendAsync("hello john doe"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); @@ -99,9 +101,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta void topicProvidedThroughContainerProperties() throws Exception { var latch = new CountDownLatch(1); var containerProps = new PulsarReaderContainerProperties(); - var config = Collections.emptyMap(); - DefaultPulsarReaderFactory pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient, config); + DefaultPulsarReaderFactory pulsarReaderFactory = new DefaultPulsarReaderFactory<>(pulsarClient); containerProps.setReaderListener((ReaderListener) (reader, msg) -> { assertThat(msg.getValue()).isEqualTo("hello buzz doe"); latch.countDown(); @@ -113,10 +114,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta try { container = new DefaultPulsarMessageReaderContainer<>(pulsarReaderFactory, containerProps); container.start(); - - Map prodConfig = Map.of("topicName", "dprlct-002"); DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>( - pulsarClient, prodConfig); + pulsarClient, "dprlct-002", (pb) -> pb.topic("dprlct-002")); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); pulsarTemplate.sendAsync("hello buzz doe"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); @@ -138,14 +137,14 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta containerProps.setTopics(List.of("dprlct-003")); containerProps.setSchema(Schema.STRING); - var readerConfig = Collections.emptyMap(); - var readerFactory = new DefaultPulsarReaderFactory(pulsarClient, readerConfig); + var readerFactory = new DefaultPulsarReaderFactory(pulsarClient); DefaultPulsarMessageReaderContainer container = null; try { container = new DefaultPulsarMessageReaderContainer<>(readerFactory, containerProps); var prodConfig = Map.of("topicName", "dprlct-003"); - var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, prodConfig); + var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, "dprlct-003", + (pb) -> pb.topic("dprlct-003")); var pulsarTemplate = new PulsarTemplate<>(producerFactory); // The following sends will not be received by the reader as we are using the 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 535ed90f..92618e72 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 @@ -19,9 +19,6 @@ package org.springframework.pulsar.reader; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -56,6 +53,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * {@link PulsarReader} integration tests. * * @author Soby Chacko + * @author Chris Bono */ @SpringJUnitConfig @DirtiesContext @@ -73,8 +71,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - Map config = Collections.emptyMap(); - return new DefaultPulsarProducerFactory<>(pulsarClient, config); + return new DefaultPulsarProducerFactory<>(pulsarClient); } @Bean @@ -89,15 +86,13 @@ public class PulsarReaderTests implements PulsarTestContainerSupport { @Bean public PulsarReaderFactory pulsarReaderFactory(PulsarClient pulsarClient) { - Map config = new HashMap<>(); - return new DefaultPulsarReaderFactory<>(pulsarClient, config); + return new DefaultPulsarReaderFactory<>(pulsarClient); } @Bean PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory pulsarReaderFactory) { - DefaultPulsarReaderContainerFactory pulsarReaderContainerFactory = new DefaultPulsarReaderContainerFactory<>( - pulsarReaderFactory, new PulsarReaderContainerProperties()); - return pulsarReaderContainerFactory; + return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory, + new PulsarReaderContainerProperties()); } }