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..a9d5b02c 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; @@ -107,7 +105,7 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient, new HashMap<>()); + return new DefaultPulsarProducerFactory<>(pulsarClient); } @Bean @@ -347,8 +345,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 +355,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 +366,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 +378,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 +490,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 +500,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 +511,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 +524,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 +658,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 +669,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-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ProducerConfigProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ProducerConfigProperties.java index e1355b84..36849b54 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ProducerConfigProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/ProducerConfigProperties.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.api.CompressionType; import org.apache.pulsar.client.api.HashingScheme; @@ -31,7 +32,7 @@ import org.apache.pulsar.client.api.ProducerCryptoFailureAction; import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.lang.Nullable; import org.springframework.pulsar.autoconfigure.PulsarProperties.Cache; -import org.springframework.pulsar.autoconfigure.PulsarProperties.Properties; +import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.util.unit.DataSize; /** @@ -365,40 +366,41 @@ public class ProducerConfigProperties { return this.cache; } - public Map buildProperties() { - PulsarProperties.Properties properties = new Properties(); - - PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); - - map.from(this::getTopicName).to(properties.in("topicName")); - map.from(this::getProducerName).to(properties.in("producerName")); - map.from(this::getSendTimeout).asInt(Duration::toMillis).to(properties.in("sendTimeoutMs")); - map.from(this::getBlockIfQueueFull).to(properties.in("blockIfQueueFull")); - map.from(this::getMaxPendingMessages).to(properties.in("maxPendingMessages")); - map.from(this::getMaxPendingMessagesAcrossPartitions).to(properties.in("maxPendingMessagesAcrossPartitions")); - map.from(this::getMessageRoutingMode).to(properties.in("messageRoutingMode")); - map.from(this::getHashingScheme).to(properties.in("hashingScheme")); - map.from(this::getCryptoFailureAction).to(properties.in("cryptoFailureAction")); - map.from(this::getBatchingMaxPublishDelay).as(it -> it.toNanos() / 1000) - .to(properties.in("batchingMaxPublishDelayMicros")); - map.from(this::getBatchingPartitionSwitchFrequencyByPublishDelay) - .to(properties.in("batchingPartitionSwitchFrequencyByPublishDelay")); - map.from(this::getBatchingMaxMessages).to(properties.in("batchingMaxMessages")); - map.from(this::getBatchingMaxBytes).asInt(DataSize::toBytes).to(properties.in("batchingMaxBytes")); - map.from(this::getBatchingEnabled).to(properties.in("batchingEnabled")); - map.from(this::getChunkingEnabled).to(properties.in("chunkingEnabled")); - map.from(this::getEncryptionKeys).to(properties.in("encryptionKeys")); - map.from(this::getCompressionType).to(properties.in("compressionType")); - map.from(this::getInitialSequenceId).to(properties.in("initialSequenceId")); - map.from(this::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions")); - map.from(this::getAutoUpdatePartitionsInterval).as(Duration::toSeconds) - .to(properties.in("autoUpdatePartitionsIntervalSeconds")); - map.from(this::getMultiSchema).to(properties.in("multiSchema")); - map.from(this::getProducerAccessMode).to(properties.in("accessMode")); - map.from(this::getLazyStartPartitionedProducers).to(properties.in("lazyStartPartitionedProducers")); - map.from(this::getProperties).to(properties.in("properties")); - - return properties; + @SuppressWarnings("deprecation") + public ProducerBuilderCustomizer toProducerBuilderCustomizer() { + return (producerBuilder) -> { + PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + map.from(this::getTopicName).to(producerBuilder::topic); + map.from(this::getProducerName).to(producerBuilder::producerName); + map.from(this::getSendTimeout).asInt(Duration::toMillis).to(producerBuilder, + (pb, val) -> pb.sendTimeout(val, TimeUnit.MILLISECONDS)); + map.from(this::getBlockIfQueueFull).to(producerBuilder::blockIfQueueFull); + map.from(this::getMaxPendingMessages).to(producerBuilder::maxPendingMessages); + map.from(this::getMaxPendingMessagesAcrossPartitions) + .to(producerBuilder::maxPendingMessagesAcrossPartitions); + map.from(this::getMessageRoutingMode).to(producerBuilder::messageRoutingMode); + map.from(this::getHashingScheme).to(producerBuilder::hashingScheme); + map.from(this::getCryptoFailureAction).to(producerBuilder::cryptoFailureAction); + map.from(this::getBatchingMaxPublishDelay).as(Duration::toMillis).to(producerBuilder, + (pb, val) -> pb.batchingMaxPublishDelay(val, TimeUnit.MILLISECONDS)); + map.from(this::getBatchingPartitionSwitchFrequencyByPublishDelay) + .to(producerBuilder::roundRobinRouterBatchingPartitionSwitchFrequency); + map.from(this::getBatchingMaxMessages).to(producerBuilder::batchingMaxMessages); + map.from(this::getBatchingMaxBytes).asInt(DataSize::toBytes).to(producerBuilder::batchingMaxBytes); + map.from(this::getBatchingEnabled).to(producerBuilder::enableBatching); + map.from(this::getChunkingEnabled).to(producerBuilder::enableChunking); + map.from(this::getEncryptionKeys) + .to((encryptionKeys) -> encryptionKeys.forEach(producerBuilder::addEncryptionKey)); + map.from(this::getCompressionType).to(producerBuilder::compressionType); + map.from(this::getInitialSequenceId).to(producerBuilder::initialSequenceId); + map.from(this::getAutoUpdatePartitions).to(producerBuilder::autoUpdatePartitions); + map.from(this::getAutoUpdatePartitionsInterval).asInt(Duration::toMillis).to(producerBuilder, + (pb, val) -> pb.autoUpdatePartitionsInterval(val, TimeUnit.MILLISECONDS)); + map.from(this::getMultiSchema).to(producerBuilder::enableMultiSchema); + map.from(this::getProducerAccessMode).to(producerBuilder::accessMode); + map.from(this::getLazyStartPartitionedProducers).to(producerBuilder::enableLazyStartPartitionedProducers); + map.from(this::getProperties).to(producerBuilder::properties); + }; } } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index e7be6bf7..7db51909 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -94,8 +94,8 @@ public class PulsarAutoConfiguration { @ConditionalOnMissingBean @ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false") public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver) { - return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(), - topicResolver); + return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(), + this.properties.getProducer().toProducerBuilderCustomizer(), topicResolver); } @Bean @@ -103,8 +103,9 @@ public class PulsarAutoConfiguration { @ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true) public PulsarProducerFactory cachingPulsarProducerFactory(PulsarClient pulsarClient, TopicResolver topicResolver) { - return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(), - topicResolver, this.properties.getProducer().getCache().getExpireAfterAccess(), + return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.getProducer().getTopicName(), + this.properties.getProducer().toProducerBuilderCustomizer(), topicResolver, + this.properties.getProducer().getCache().getExpireAfterAccess(), this.properties.getProducer().getCache().getMaximumSize(), this.properties.getProducer().getCache().getInitialCapacity()); } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index 8123c3dc..c4dbc82a 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java @@ -110,10 +110,6 @@ public class PulsarProperties { return new HashMap<>(this.consumer.buildProperties()); } - public Map buildProducerProperties() { - return new HashMap<>(this.producer.buildProperties()); - } - public Map buildAdminProperties() { return new HashMap<>(this.admin.buildProperties()); } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index a69500ee..0e9dae9e 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -22,6 +22,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.util.Collections; import java.util.List; @@ -42,6 +43,7 @@ import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.test.context.FilteredClassLoader; +import org.springframework.boot.test.context.TestConfiguration; import org.springframework.boot.test.context.assertj.AssertableApplicationContext; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.context.annotation.Bean; @@ -58,6 +60,7 @@ import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.DefaultPulsarReaderFactory; import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.DefaultTopicResolver; +import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarClientBuilderCustomizer; import org.springframework.pulsar.core.PulsarConsumerFactory; @@ -369,10 +372,8 @@ class PulsarAutoConfigurationTests { @Test void clientConfigurerWithNoUserDefinedCustomizers() { - contextRunner.run((context) -> { - assertThat(context).getBean(PulsarClientBuilderConfigurer.class) - .hasFieldOrPropertyWithValue("customizers", Collections.emptyList()); - }); + contextRunner.run((context) -> assertThat(context).getBean(PulsarClientBuilderConfigurer.class) + .hasFieldOrPropertyWithValue("customizers", Collections.emptyList())); } @Test @@ -520,18 +521,28 @@ class PulsarAutoConfigurationTests { @Test void beansAreInjectedInNonCachingProducerFactory() { - contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=false") - .run((context -> assertThat(context).hasNotFailed().getBean(DefaultPulsarProducerFactory.class) + contextRunner.withUserConfiguration(ProducerCustomizerConfig.class) + .withPropertyValues("spring.pulsar.producer.topic-name=foo-topic", + "spring.pulsar.producer.cache.enabled=false") + .run((context) -> assertThat(context).getBean(DefaultPulsarProducerFactory.class) + .hasFieldOrPropertyWithValue("defaultTopic", "foo-topic") + .hasFieldOrPropertyWithValue("defaultConfigCustomizer", + ProducerCustomizerConfig.testCustomizer) .hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)) - .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class)))); + .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))); } @Test void beansAreInjectedInCachingProducerFactory() { - contextRunner.withPropertyValues("spring.pulsar.producer.cache.enabled=true") - .run((context -> assertThat(context).hasNotFailed().getBean(CachingPulsarProducerFactory.class) + contextRunner.withUserConfiguration(ProducerCustomizerConfig.class) + .withPropertyValues("spring.pulsar.producer.topic-name=foo-topic", + "spring.pulsar.producer.cache.enabled=true") + .run((context) -> assertThat(context).getBean(CachingPulsarProducerFactory.class) + .hasFieldOrPropertyWithValue("defaultTopic", "foo-topic") + .hasFieldOrPropertyWithValue("defaultConfigCustomizer", + ProducerCustomizerConfig.testCustomizer) .hasFieldOrPropertyWithValue("pulsarClient", context.getBean(PulsarClient.class)) - .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class)))); + .hasFieldOrPropertyWithValue("topicResolver", context.getBean(TopicResolver.class))); } private void assertHasProducerFactoryOfType(Class producerFactoryType, @@ -610,4 +621,29 @@ class PulsarAutoConfigurationTests { } + /* + * Use '@TestConfiguration' and exact name of the PulsarProperties bean that is + * created via the '@EnableConfigurationProperties' on the actual auto-config in order + * to 'replace' the PulsarProperties bean - all so we can make sure the returned + * producer builder customizer is the one we expect. + */ + @TestConfiguration(proxyBeanMethods = false) + static class ProducerCustomizerConfig { + + @SuppressWarnings("rawtypes") + static ProducerBuilderCustomizer testCustomizer = (producerBuilder) -> { + }; + + @Bean(name = "spring.pulsar-org.springframework.pulsar.autoconfigure.PulsarProperties") + PulsarProperties pulsarProperties() { + var pulsarProps = new PulsarProperties(); + var producerProps = spy(pulsarProps.getProducer()); + when(producerProps.toProducerBuilderCustomizer()).thenReturn(testCustomizer); + var spyPulsarProps = spy(pulsarProps); + when(spyPulsarProps.getProducer()).thenReturn(producerProps); + return spyPulsarProps; + } + + } + } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java index 06eba955..3315fd90 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatRuntimeException; +import static org.assertj.core.api.Assertions.entry; import java.time.Duration; import java.util.HashMap; @@ -41,7 +42,6 @@ import org.apache.pulsar.client.api.SubscriptionMode; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils; import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; -import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; import org.apache.pulsar.common.schema.SchemaType; import org.assertj.core.api.InstanceOfAssertFactories; @@ -58,7 +58,7 @@ import org.springframework.pulsar.autoconfigure.PulsarProperties.TypeMapping; import org.springframework.util.unit.DataSize; /** - * Unit tests for {@link PulsarProperties}. + * Tests for {@link PulsarProperties}. * * @author Chris Bono * @author Christophe Bornet @@ -389,7 +389,7 @@ public class PulsarPropertiesTests { @Test void producerProperties() { - Map props = new HashMap<>(); + var props = new HashMap(); props.put("spring.pulsar.producer.topic-name", "my-topic"); props.put("spring.pulsar.producer.producer-name", "my-producer"); props.put("spring.pulsar.producer.send-timeout", "2s"); @@ -413,32 +413,29 @@ public class PulsarPropertiesTests { props.put("spring.pulsar.producer.properties[my-prop]", "my-prop-value"); bind(props); - Map producerProps = properties.buildProducerProperties(); - // Verify that the props can be loaded in a ProducerBuilder - assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps, - new ProducerConfigurationData(), ProducerConfigurationData.class)); - - assertThat(producerProps).containsEntry("topicName", "my-topic") - .containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000) - .containsEntry("blockIfQueueFull", true).containsEntry("maxPendingMessages", 3) - .containsEntry("maxPendingMessagesAcrossPartitions", 4) - .containsEntry("messageRoutingMode", MessageRoutingMode.CustomPartition) - .containsEntry("hashingScheme", HashingScheme.Murmur3_32Hash) - .containsEntry("cryptoFailureAction", ProducerCryptoFailureAction.SEND) - .containsEntry("batchingMaxPublishDelayMicros", 5_000_000L) - .containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6) - .containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8) - .containsEntry("batchingEnabled", false).containsEntry("chunkingEnabled", true) - .hasEntrySatisfying("encryptionKeys", - keys -> assertThat(keys).asInstanceOf(InstanceOfAssertFactories.collection(String.class)) - .containsExactly("my-key")) - .containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L) - .containsEntry("accessMode", ProducerAccessMode.Exclusive) - .containsEntry("lazyStartPartitionedProducers", true).hasEntrySatisfying("properties", - properties -> assertThat(properties) - .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) - .containsEntry("my-prop", "my-prop-value")); + var producerProps = properties.getProducer(); + assertThat(producerProps.getTopicName()).isEqualTo("my-topic"); + assertThat(producerProps.getProducerName()).isEqualTo("my-producer"); + assertThat(producerProps.getSendTimeout()).isEqualTo(Duration.ofMillis(2000)); + assertThat(producerProps.getBlockIfQueueFull()).isTrue(); + assertThat(producerProps.getMaxPendingMessages()).isEqualTo(3); + assertThat(producerProps.getMaxPendingMessagesAcrossPartitions()).isEqualTo(4); + assertThat(producerProps.getMessageRoutingMode()).isEqualTo(MessageRoutingMode.CustomPartition); + assertThat(producerProps.getHashingScheme()).isEqualTo(HashingScheme.Murmur3_32Hash); + assertThat(producerProps.getCryptoFailureAction()).isEqualTo(ProducerCryptoFailureAction.SEND); + assertThat(producerProps.getBatchingMaxPublishDelay()).isEqualTo(Duration.ofMillis(5000)); + assertThat(producerProps.getBatchingPartitionSwitchFrequencyByPublishDelay()).isEqualTo(6); + assertThat(producerProps.getBatchingMaxMessages()).isEqualTo(7); + assertThat(producerProps.getBatchingMaxBytes()).isEqualTo(DataSize.ofBytes(8)); + assertThat(producerProps.getBatchingEnabled()).isFalse(); + assertThat(producerProps.getChunkingEnabled()).isTrue(); + assertThat(producerProps.getEncryptionKeys()).containsExactly("my-key"); + assertThat(producerProps.getCompressionType()).isEqualTo(CompressionType.LZ4); + assertThat(producerProps.getInitialSequenceId()).isEqualTo(9); + assertThat(producerProps.getProducerAccessMode()).isEqualTo(ProducerAccessMode.Exclusive); + assertThat(producerProps.getLazyStartPartitionedProducers()).isTrue(); + assertThat(producerProps.getProperties()).containsExactly(entry("my-prop", "my-prop-value")); } } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java index cef8b8c1..8acd5e89 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java @@ -16,15 +16,19 @@ package org.springframework.pulsar.spring.cloud.stream.binder; +import java.time.Duration; import java.util.HashMap; import java.util.Map; import java.util.Objects; import java.util.UUID; +import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.autoconfigure.ProducerConfigProperties; import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; import org.springframework.util.StringUtils; +import org.springframework.util.unit.DataSize; /** * Binder utility methods. @@ -100,4 +104,52 @@ final class PulsarBinderUtils { return newOrModifiedProps; } + /** + * Gets a map representation of a {@link ProducerConfigProperties}. + * @param producerProps the producer props + * @return map representation of producer props where each entry is a field and its + * associated value + */ + static Map convertProducerPropertiesToMap(ProducerConfigProperties producerProps) { + var properties = new PulsarBinderUtils.Properties(); + var map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + map.from(producerProps::getTopicName).to(properties.in("topicName")); + map.from(producerProps::getProducerName).to(properties.in("producerName")); + map.from(producerProps::getSendTimeout).asInt(Duration::toMillis).to(properties.in("sendTimeoutMs")); + map.from(producerProps::getBlockIfQueueFull).to(properties.in("blockIfQueueFull")); + map.from(producerProps::getMaxPendingMessages).to(properties.in("maxPendingMessages")); + map.from(producerProps::getMaxPendingMessagesAcrossPartitions) + .to(properties.in("maxPendingMessagesAcrossPartitions")); + map.from(producerProps::getMessageRoutingMode).to(properties.in("messageRoutingMode")); + map.from(producerProps::getHashingScheme).to(properties.in("hashingScheme")); + map.from(producerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction")); + map.from(producerProps::getBatchingMaxPublishDelay).as(it -> it.toNanos() / 1000) + .to(properties.in("batchingMaxPublishDelayMicros")); + map.from(producerProps::getBatchingPartitionSwitchFrequencyByPublishDelay) + .to(properties.in("batchingPartitionSwitchFrequencyByPublishDelay")); + map.from(producerProps::getBatchingMaxMessages).to(properties.in("batchingMaxMessages")); + map.from(producerProps::getBatchingMaxBytes).asInt(DataSize::toBytes).to(properties.in("batchingMaxBytes")); + map.from(producerProps::getBatchingEnabled).to(properties.in("batchingEnabled")); + map.from(producerProps::getChunkingEnabled).to(properties.in("chunkingEnabled")); + map.from(producerProps::getEncryptionKeys).to(properties.in("encryptionKeys")); + map.from(producerProps::getCompressionType).to(properties.in("compressionType")); + map.from(producerProps::getInitialSequenceId).to(properties.in("initialSequenceId")); + map.from(producerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions")); + map.from(producerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds) + .to(properties.in("autoUpdatePartitionsIntervalSeconds")); + map.from(producerProps::getMultiSchema).to(properties.in("multiSchema")); + map.from(producerProps::getProducerAccessMode).to(properties.in("accessMode")); + map.from(producerProps::getLazyStartPartitionedProducers).to(properties.in("lazyStartPartitionedProducers")); + map.from(producerProps::getProperties).to(properties.in("properties")); + return properties; + } + + static class Properties extends HashMap { + + java.util.function.Consumer in(String key) { + return (value) -> put(key, value); + } + + } + } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java index 288c733f..c31e3f81 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java @@ -110,9 +110,11 @@ public class PulsarMessageChannelBinder extends else { schema = null; } - var baseProducerProps = new ProducerConfigProperties().buildProperties(); - var binderProducerProps = this.binderConfigProps.getProducer().buildProperties(); - var bindingProducerProps = producerProperties.getExtension().buildProperties(); + + var baseProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(new ProducerConfigProperties()); + var binderProducerProps = PulsarBinderUtils + .convertProducerPropertiesToMap(this.binderConfigProps.getProducer()); + var bindingProducerProps = PulsarBinderUtils.convertProducerPropertiesToMap(producerProperties.getExtension()); var mergedProducerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseProducerProps, binderProducerProps, bindingProducerProps); diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java index a51750d4..7ff3c387 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderConfigurationPropertiesTests.java @@ -68,7 +68,7 @@ public class PulsarBinderConfigurationPropertiesTests { props.put("spring.cloud.stream.pulsar.binder.producer.properties[my-prop]", "my-prop-value"); bind(props); - Map producerProps = properties.getProducer().buildProperties(); + Map producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties.getProducer()); // Verify that the props can be loaded in a ProducerBuilder assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps, diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java index a4242985..f8d41366 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java @@ -639,10 +639,14 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { @Import(PrimitiveTextConfig.class) static class BinderAndBindingPropsTestConfig { + @SuppressWarnings("unchecked") @Bean public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient, PulsarProperties pulsarProperties, TopicResolver topicResolver) { - return new TrackingProducerFactory(pulsarClient, pulsarProperties.buildProducerProperties(), topicResolver); + var customizer = (ProducerBuilderCustomizer) pulsarProperties.getProducer() + .toProducerBuilderCustomizer(); + return new TrackingProducerFactory(pulsarClient, pulsarProperties.getProducer().getTopicName(), customizer, + topicResolver); } @Bean @@ -657,8 +661,9 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { List> producersCreated = new ArrayList<>(); - TrackingProducerFactory(PulsarClient pulsarClient, Map config, TopicResolver topicResolver) { - super(pulsarClient, config, topicResolver); + TrackingProducerFactory(PulsarClient pulsarClient, @Nullable String defaultTopic, + ProducerBuilderCustomizer defaultConfigCustomizer, TopicResolver topicResolver) { + super(pulsarClient, defaultTopic, defaultConfigCustomizer, topicResolver); } @Override diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java index 630ddfb2..49b07b3a 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java @@ -19,7 +19,6 @@ package org.springframework.pulsar.spring.cloud.stream.binder; import static org.assertj.core.api.Assertions.assertThat; import java.nio.charset.StandardCharsets; -import java.util.Collections; import java.util.Map; import java.util.UUID; import java.util.concurrent.CountDownLatch; @@ -105,7 +104,7 @@ public class PulsarBinderTests extends Map.of("serviceUrl", PulsarTestContainerSupport.getHttpServiceUrl())); var configProps = new PulsarBinderConfigurationProperties(); var provisioner = new PulsarTopicProvisioner(pulsarAdministration, configProps); - var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap()); + var producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient); var pulsarTemplate = new PulsarTemplate<>(producerFactory); var config = Map.of("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest); var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, config); diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java index 8155a3be..95d2839c 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java @@ -17,21 +17,36 @@ package org.springframework.pulsar.spring.cloud.stream.binder; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatNoException; import static org.junit.jupiter.params.provider.Arguments.arguments; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; import java.util.Collections; +import java.util.HashMap; import java.util.Map; import java.util.stream.Stream; +import org.apache.pulsar.client.api.CompressionType; +import org.apache.pulsar.client.api.HashingScheme; +import org.apache.pulsar.client.api.MessageRoutingMode; +import org.apache.pulsar.client.api.ProducerAccessMode; +import org.apache.pulsar.client.api.ProducerCryptoFailureAction; +import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils; +import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; +import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.springframework.boot.context.properties.bind.Bindable; +import org.springframework.boot.context.properties.bind.Binder; +import org.springframework.boot.context.properties.source.ConfigurationPropertySource; +import org.springframework.boot.context.properties.source.MapConfigurationPropertySource; import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.pulsar.autoconfigure.ProducerConfigProperties; import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; /** @@ -130,4 +145,70 @@ public class PulsarBinderUtilsTests { } + @Nested + class ConvertedPropertiesTests { + + private final ProducerConfigProperties properties = new ProducerConfigProperties(); + + private void bind(Map map) { + ConfigurationPropertySource source = new MapConfigurationPropertySource(map); + new Binder(source).bind("spring.pulsar.producer", Bindable.ofInstance(this.properties)); + } + + @Test + void producerPropertiesToMap() { + Map props = new HashMap<>(); + props.put("spring.pulsar.producer.topic-name", "my-topic"); + props.put("spring.pulsar.producer.producer-name", "my-producer"); + props.put("spring.pulsar.producer.send-timeout", "2s"); + props.put("spring.pulsar.producer.block-if-queue-full", "true"); + props.put("spring.pulsar.producer.max-pending-messages", "3"); + props.put("spring.pulsar.producer.max-pending-messages-across-partitions", "4"); + props.put("spring.pulsar.producer.message-routing-mode", "custompartition"); + props.put("spring.pulsar.producer.hashing-scheme", "murmur3_32hash"); + props.put("spring.pulsar.producer.crypto-failure-action", "send"); + props.put("spring.pulsar.producer.batching-max-publish-delay", "5s"); + props.put("spring.pulsar.producer.batching-partition-switch-frequency-by-publish-delay", "6"); + props.put("spring.pulsar.producer.batching-max-messages", "7"); + props.put("spring.pulsar.producer.batching-max-bytes", "8"); + props.put("spring.pulsar.producer.batching-enabled", "false"); + props.put("spring.pulsar.producer.chunking-enabled", "true"); + props.put("spring.pulsar.producer.encryption-keys[0]", "my-key"); + props.put("spring.pulsar.producer.compression-type", "lz4"); + props.put("spring.pulsar.producer.initial-sequence-id", "9"); + props.put("spring.pulsar.producer.producer-access-mode", "exclusive"); + props.put("spring.pulsar.producer.lazy-start=partitioned-producers", "true"); + props.put("spring.pulsar.producer.properties[my-prop]", "my-prop-value"); + + bind(props); + Map producerProps = PulsarBinderUtils.convertProducerPropertiesToMap(properties); + + // Verify that the props can be loaded in a ProducerBuilder + assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps, + new ProducerConfigurationData(), ProducerConfigurationData.class)); + + assertThat(producerProps).containsEntry("topicName", "my-topic") + .containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000) + .containsEntry("blockIfQueueFull", true).containsEntry("maxPendingMessages", 3) + .containsEntry("maxPendingMessagesAcrossPartitions", 4) + .containsEntry("messageRoutingMode", MessageRoutingMode.CustomPartition) + .containsEntry("hashingScheme", HashingScheme.Murmur3_32Hash) + .containsEntry("cryptoFailureAction", ProducerCryptoFailureAction.SEND) + .containsEntry("batchingMaxPublishDelayMicros", 5_000_000L) + .containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6) + .containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8) + .containsEntry("batchingEnabled", false).containsEntry("chunkingEnabled", true) + .hasEntrySatisfying("encryptionKeys", + keys -> assertThat(keys).asInstanceOf(InstanceOfAssertFactories.collection(String.class)) + .containsExactly("my-key")) + .containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L) + .containsEntry("accessMode", ProducerAccessMode.Exclusive) + .containsEntry("lazyStartPartitionedProducers", true).hasEntrySatisfying("properties", + properties -> assertThat(properties) + .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) + .containsEntry("my-prop", "my-prop-value")); + } + + } + } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java index 3a2e56a5..5abf6254 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarExtendedBindingPropertiesTests.java @@ -65,7 +65,8 @@ public class PulsarExtendedBindingPropertiesTests { bind(props); assertThat(properties.getBindings()).containsOnlyKeys("my-foo"); - Map producerProps = properties.getExtendedProducerProperties("my-foo").buildProperties(); + Map producerProps = PulsarBinderUtils + .convertProducerPropertiesToMap(properties.getExtendedProducerProperties("my-foo")); // Verify that the props can be loaded in a ProducerBuilder assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps, new ProducerConfigurationData(), ProducerConfigurationData.class)); 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/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/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..a6a751dd 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 @@ -85,9 +85,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"); @@ -115,9 +114,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"); @@ -160,9 +158,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"); @@ -226,9 +223,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"); @@ -272,9 +268,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { 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"); @@ -316,9 +311,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { 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"); @@ -352,9 +346,8 @@ class ConsumerAcknowledgmentTests implements PulsarTestContainerSupport { 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"); 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..cb624ca2 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; @@ -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"); 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..f6804843 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 @@ -86,10 +86,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/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..9fd94273 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 @@ -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(); @@ -132,7 +131,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..e5b6461e 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 @@ -74,9 +74,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); @@ -127,9 +126,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); @@ -174,9 +172,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); @@ -252,9 +249,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); @@ -321,9 +317,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); @@ -389,9 +384,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); @@ -463,9 +457,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); @@ -533,9 +526,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..3ff1daf2 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 @@ -81,9 +81,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(); @@ -178,9 +177,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); @@ -209,9 +207,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); @@ -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); @@ -312,9 +308,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); @@ -371,9 +366,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..86321486 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,10 +22,8 @@ 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; @@ -105,8 +103,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,16 +118,14 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { - Map config = new HashMap<>(); - return new DefaultPulsarConsumerFactory<>(pulsarClient, config); + return new DefaultPulsarConsumerFactory<>(pulsarClient, new HashMap<>()); } @Bean PulsarListenerContainerFactory pulsarListenerContainerFactory( PulsarConsumerFactory pulsarConsumerFactory) { - ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>( - pulsarConsumerFactory, new PulsarContainerProperties()); - return pulsarListenerContainerFactory; + return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, + new PulsarContainerProperties()); } @Bean @@ -184,12 +179,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 +196,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 +451,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 +463,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 +475,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 +487,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 +626,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 +637,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 +648,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 +660,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 +730,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 +741,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); } 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..35c85e96 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 @@ -131,7 +131,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 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..5bdf0bd9 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 @@ -170,7 +170,7 @@ public class ObservationTests implements PulsarTestContainerSupport { @Bean PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient, Collections.emptyMap()); + return new DefaultPulsarProducerFactory<>(pulsarClient); } @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..4505c977 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 @@ -44,6 +44,7 @@ import org.springframework.pulsar.test.support.PulsarTestContainerSupport; * Basic tests for {@link DefaultPulsarMessageReaderContainer}. * * @author Soby Chacko + * @author Chris Bono */ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestContainerSupport { @@ -83,9 +84,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(); @@ -113,10 +113,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(); @@ -145,7 +143,8 @@ public class DefaultPulsarMessageReaderContainerTests implements PulsarTestConta 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..63f55af8 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,7 @@ 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 +54,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * {@link PulsarReader} integration tests. * * @author Soby Chacko + * @author Chris Bono */ @SpringJUnitConfig @DirtiesContext @@ -73,8 +72,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 +87,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, new HashMap<>()); } @Bean PulsarReaderContainerFactory pulsarReaderContainerFactory(PulsarReaderFactory pulsarReaderFactory) { - DefaultPulsarReaderContainerFactory pulsarReaderContainerFactory = new DefaultPulsarReaderContainerFactory<>( - pulsarReaderFactory, new PulsarReaderContainerProperties()); - return pulsarReaderContainerFactory; + return new DefaultPulsarReaderContainerFactory<>(pulsarReaderFactory, + new PulsarReaderContainerProperties()); } }