diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java index 51e6e18cd..229372d7c 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtils.java @@ -16,16 +16,20 @@ package org.springframework.cloud.stream.binder.pulsar; +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.binder.pulsar.properties.PulsarConsumerProperties; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties; +import org.springframework.pulsar.autoconfigure.ProducerConfigProperties; import org.springframework.util.StringUtils; - +import org.springframework.util.unit.DataSize; /** * Binder utility methods. @@ -101,4 +105,103 @@ 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; + } + + /** + * Gets a map representation of a {@link ConsumerConfigProperties}. + * @param consumerProps the consumer props + * @return map representation of consumer props where each entry is a field and its + * associated value + */ + static Map convertConsumerPropertiesToMap(ConsumerConfigProperties consumerProps) { + var properties = new PulsarBinderUtils.Properties(); + var map = PropertyMapper.get().alwaysApplyingWhenNonNull(); + map.from(consumerProps::getTopics).to(properties.in("topicNames")); + map.from(consumerProps::getTopicsPattern).to(properties.in("topicsPattern")); + map.from(consumerProps::getSubscriptionName).to(properties.in("subscriptionName")); + map.from(consumerProps::getSubscriptionType).to(properties.in("subscriptionType")); + map.from(consumerProps::getSubscriptionProperties).to(properties.in("subscriptionProperties")); + map.from(consumerProps::getSubscriptionMode).to(properties.in("subscriptionMode")); + map.from(consumerProps::getReceiverQueueSize).to(properties.in("receiverQueueSize")); + map.from(consumerProps::getAcknowledgementsGroupTime).as(it -> it.toNanos() / 1000) + .to(properties.in("acknowledgementsGroupTimeMicros")); + map.from(consumerProps::getNegativeAckRedeliveryDelay).as(it -> it.toNanos() / 1000) + .to(properties.in("negativeAckRedeliveryDelayMicros")); + map.from(consumerProps::getMaxTotalReceiverQueueSizeAcrossPartitions) + .to(properties.in("maxTotalReceiverQueueSizeAcrossPartitions")); + map.from(consumerProps::getConsumerName).to(properties.in("consumerName")); + map.from(consumerProps::getAckTimeout).as(Duration::toMillis).to(properties.in("ackTimeoutMillis")); + map.from(consumerProps::getTickDuration).as(Duration::toMillis).to(properties.in("tickDurationMillis")); + map.from(consumerProps::getPriorityLevel).to(properties.in("priorityLevel")); + map.from(consumerProps::getCryptoFailureAction).to(properties.in("cryptoFailureAction")); + map.from(consumerProps::getProperties).to(properties.in("properties")); + map.from(consumerProps::getReadCompacted).to(properties.in("readCompacted")); + map.from(consumerProps::getSubscriptionInitialPosition).to(properties.in("subscriptionInitialPosition")); + map.from(consumerProps::getPatternAutoDiscoveryPeriod).to(properties.in("patternAutoDiscoveryPeriod")); + map.from(consumerProps::getRegexSubscriptionMode).to(properties.in("regexSubscriptionMode")); + map.from(consumerProps::getDeadLetterPolicy).to(properties.in("deadLetterPolicy")); + map.from(consumerProps::getRetryEnable).to(properties.in("retryEnable")); + map.from(consumerProps::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions")); + map.from(consumerProps::getAutoUpdatePartitionsInterval).as(Duration::toSeconds) + .to(properties.in("autoUpdatePartitionsIntervalSeconds")); + map.from(consumerProps::getReplicateSubscriptionState).to(properties.in("replicateSubscriptionState")); + map.from(consumerProps::getResetIncludeHead).to(properties.in("resetIncludeHead")); + map.from(consumerProps::getBatchIndexAckEnabled).to(properties.in("batchIndexAckEnabled")); + map.from(consumerProps::getAckReceiptEnabled).to(properties.in("ackReceiptEnabled")); + map.from(consumerProps::getPoolMessages).to(properties.in("poolMessages")); + map.from(consumerProps::getStartPaused).to(properties.in("startPaused")); + map.from(consumerProps::getAutoAckOldestChunkedMessageOnQueueFull) + .to(properties.in("autoAckOldestChunkedMessageOnQueueFull")); + map.from(consumerProps::getMaxPendingChunkedMessage).to(properties.in("maxPendingChunkedMessage")); + map.from(consumerProps::getExpireTimeOfIncompleteChunkedMessage).as(Duration::toMillis) + .to(properties.in("expireTimeOfIncompleteChunkedMessageMillis")); + return properties; + } + + static class Properties extends HashMap { + + java.util.function.Consumer in(String key) { + return (value) -> put(key, value); + } + + } + } diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java index 16de5c11c..d6eafe2eb 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/main/java/org/springframework/cloud/stream/binder/pulsar/PulsarMessageChannelBinder.java @@ -111,9 +111,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); @@ -165,9 +167,10 @@ public class PulsarMessageChannelBinder extends var subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination); containerProperties.setSubscriptionName(subscriptionName); - var baseConsumerProps = new ConsumerConfigProperties().buildProperties(); - var binderConsumerProps = this.binderConfigProps.getConsumer().buildProperties(); - var bindingConsumerProps = properties.getExtension().buildProperties(); + var baseConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(new ConsumerConfigProperties()); + var binderConsumerProps = PulsarBinderUtils + .convertConsumerPropertiesToMap(this.binderConfigProps.getConsumer()); + var bindingConsumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getExtension()); var mergedConsumerProps = PulsarBinderUtils.mergePropertiesWithPrecedence(baseConsumerProps, binderConsumerProps, bindingConsumerProps); containerProperties.getPulsarConsumerProperties().putAll(mergedConsumerProps); diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderConfigurationPropertiesTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderConfigurationPropertiesTests.java index a1b8cdffa..d21b40335 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderConfigurationPropertiesTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/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, @@ -98,7 +98,7 @@ public class PulsarBinderConfigurationPropertiesTests { props.put("spring.cloud.stream.pulsar.binder.consumer.receiver-queue-size", "1"); bind(props); - Map consumerProps = properties.getConsumer().buildProperties(); + Map consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties.getConsumer()); // Verify that the props can be loaded in a ConsumerBuilder assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java index c0030cf41..08ec2731b 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderIntegrationTests.java @@ -638,16 +638,23 @@ 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); } + @SuppressWarnings("unchecked") @Bean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, PulsarProperties pulsarProperties) { - return new TrackingConsumerFactory(pulsarClient, pulsarProperties.buildConsumerProperties()); + var customizer = (ConsumerBuilderCustomizer) pulsarProperties.getConsumer() + .toConsumerBuilderCustomizer(); + return new TrackingConsumerFactory(pulsarClient, customizer); } } @@ -656,8 +663,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 @@ -677,8 +685,8 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { List> consumersCreated = new ArrayList<>(); - TrackingConsumerFactory(PulsarClient pulsarClient, Map consumerConfig) { - super(pulsarClient, consumerConfig); + TrackingConsumerFactory(PulsarClient pulsarClient, ConsumerBuilderCustomizer defaultConsumerConfig) { + super(pulsarClient, defaultConsumerConfig); } @Override diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java index 100b75138..a412d3890 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderTests.java @@ -17,8 +17,6 @@ package org.springframework.cloud.stream.binder.pulsar; import java.nio.charset.StandardCharsets; -import java.util.Collections; -import java.util.Map; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -100,14 +98,13 @@ public class PulsarBinderTests extends @Override protected PulsarTestBinder getBinder() { - var pulsarAdministration = new PulsarAdministration( - Map.of("serviceUrl", PulsarTestContainerSupport.getHttpServiceUrl())); + var pulsarAdministration = new PulsarAdministration(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); + var consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, + (consumerBuilder -> consumerBuilder.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest))); if (this.binder == null) { this.binder = new PulsarTestBinder(provisioner, pulsarTemplate, consumerFactory, configProps, new DefaultSchemaResolver(), JsonPulsarHeaderMapper.builder().build()); diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java index c23a425d8..17805d225 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarBinderUtilsTests.java @@ -17,19 +17,42 @@ package org.springframework.cloud.stream.binder.pulsar; 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.ConsumerCryptoFailureAction; +import org.apache.pulsar.client.api.DeadLetterPolicy; +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.api.RegexSubscriptionMode; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +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.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.binder.pulsar.properties.PulsarConsumerProperties; import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.pulsar.autoconfigure.ConsumerConfigProperties; +import org.springframework.pulsar.autoconfigure.ProducerConfigProperties; 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; @@ -130,4 +153,172 @@ 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")); + } + + } + + @Nested + class ConvertedConsumerPropertiesTests { + + private final ConsumerConfigProperties properties = new ConsumerConfigProperties(); + + private void bind(Map map) { + ConfigurationPropertySource source = new MapConfigurationPropertySource(map); + new Binder(source).bind("spring.pulsar.consumer", Bindable.ofInstance(this.properties)); + } + + @Test + void consumerPropertiesToMap() { + Map props = new HashMap<>(); + props.put("spring.pulsar.consumer.topics[0]", "my-topic"); + props.put("spring.pulsar.consumer.topics-pattern", "my-pattern"); + props.put("spring.pulsar.consumer.subscription-name", "my-subscription"); + props.put("spring.pulsar.consumer.subscription-type", "shared"); + props.put("spring.pulsar.consumer.subscription-properties[my-sub-prop]", "my-sub-prop-value"); + props.put("spring.pulsar.consumer.subscription-mode", "nondurable"); + props.put("spring.pulsar.consumer.receiver-queue-size", "1"); + props.put("spring.pulsar.consumer.acknowledgements-group-time", "2s"); + props.put("spring.pulsar.consumer.negative-ack-redelivery-delay", "3s"); + props.put("spring.pulsar.consumer.max-total-receiver-queue-size-across-partitions", "5"); + props.put("spring.pulsar.consumer.consumer-name", "my-consumer"); + props.put("spring.pulsar.consumer.ack-timeout", "6s"); + props.put("spring.pulsar.consumer.tick-duration", "7s"); + props.put("spring.pulsar.consumer.priority-level", "8"); + props.put("spring.pulsar.consumer.crypto-failure-action", "discard"); + props.put("spring.pulsar.consumer.properties[my-prop]", "my-prop-value"); + props.put("spring.pulsar.consumer.read-compacted", "true"); + props.put("spring.pulsar.consumer.subscription-initial-position", "earliest"); + props.put("spring.pulsar.consumer.pattern-auto-discovery-period", "9"); + props.put("spring.pulsar.consumer.regex-subscription-mode", "all-topics"); + props.put("spring.pulsar.consumer.dead-letter-policy.max-redeliver-count", "4"); + props.put("spring.pulsar.consumer.dead-letter-policy.retry-letter-topic", "my-retry-topic"); + props.put("spring.pulsar.consumer.dead-letter-policy.dead-letter-topic", "my-dlt-topic"); + props.put("spring.pulsar.consumer.dead-letter-policy.initial-subscription-name", "my-initial-subscription"); + props.put("spring.pulsar.consumer.retry-enable", "true"); + props.put("spring.pulsar.consumer.auto-update-partitions", "false"); + props.put("spring.pulsar.consumer.auto-update-partitions-interval", "10s"); + props.put("spring.pulsar.consumer.replicate-subscription-state", "true"); + props.put("spring.pulsar.consumer.reset-include-head", "true"); + props.put("spring.pulsar.consumer.batch-index-ack-enabled", "true"); + props.put("spring.pulsar.consumer.ack-receipt-enabled", "true"); + props.put("spring.pulsar.consumer.pool-messages", "true"); + props.put("spring.pulsar.consumer.start-paused", "true"); + props.put("spring.pulsar.consumer.auto-ack-oldest-chunked-message-on-queue-full", "false"); + props.put("spring.pulsar.consumer.max-pending-chunked-message", "11"); + props.put("spring.pulsar.consumer.expire-time-of-incomplete-chunked-message", "12s"); + + bind(props); + Map consumerProps = PulsarBinderUtils.convertConsumerPropertiesToMap(properties); + + // Verify that the props can be loaded in a ConsumerBuilder + assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, + new ConsumerConfigurationData<>(), ConsumerConfigurationData.class)); + + assertThat(consumerProps) + .hasEntrySatisfying("topicNames", + topics -> assertThat(topics) + .asInstanceOf(InstanceOfAssertFactories.collection(String.class)) + .containsExactly("my-topic")) + .hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern")) + .containsEntry("subscriptionName", "my-subscription") + .containsEntry("subscriptionType", SubscriptionType.Shared) + .hasEntrySatisfying("subscriptionProperties", + properties -> assertThat(properties) + .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) + .containsEntry("my-sub-prop", "my-sub-prop-value")) + .containsEntry("subscriptionMode", SubscriptionMode.NonDurable) + .containsEntry("receiverQueueSize", 1).containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L) + .containsEntry("negativeAckRedeliveryDelayMicros", 3_000_000L) + .containsEntry("maxTotalReceiverQueueSizeAcrossPartitions", 5) + .containsEntry("consumerName", "my-consumer").containsEntry("ackTimeoutMillis", 6_000L) + .containsEntry("tickDurationMillis", 7_000L).containsEntry("priorityLevel", 8) + .containsEntry("cryptoFailureAction", ConsumerCryptoFailureAction.DISCARD) + .hasEntrySatisfying("properties", + properties -> assertThat(properties) + .asInstanceOf(InstanceOfAssertFactories.map(String.class, String.class)) + .containsEntry("my-prop", "my-prop-value")) + .containsEntry("readCompacted", true) + .containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest) + .containsEntry("patternAutoDiscoveryPeriod", 9) + .containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics) + .hasEntrySatisfying("deadLetterPolicy", dlp -> { + DeadLetterPolicy deadLetterPolicy = (DeadLetterPolicy) dlp; + assertThat(deadLetterPolicy.getMaxRedeliverCount()).isEqualTo(4); + assertThat(deadLetterPolicy.getRetryLetterTopic()).isEqualTo("my-retry-topic"); + assertThat(deadLetterPolicy.getDeadLetterTopic()).isEqualTo("my-dlt-topic"); + assertThat(deadLetterPolicy.getInitialSubscriptionName()).isEqualTo("my-initial-subscription"); + }).containsEntry("retryEnable", true).containsEntry("autoUpdatePartitions", false) + .containsEntry("autoUpdatePartitionsIntervalSeconds", 10L) + .containsEntry("replicateSubscriptionState", true).containsEntry("resetIncludeHead", true) + .containsEntry("batchIndexAckEnabled", true).containsEntry("ackReceiptEnabled", true) + .containsEntry("poolMessages", true).containsEntry("startPaused", true) + .containsEntry("autoAckOldestChunkedMessageOnQueueFull", false) + .containsEntry("maxPendingChunkedMessage", 11) + .containsEntry("expireTimeOfIncompleteChunkedMessageMillis", 12_000L); + } + + } + } diff --git a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingPropertiesTests.java b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingPropertiesTests.java index 38a68a729..c272e064b 100644 --- a/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/PulsarExtendedBindingPropertiesTests.java +++ b/binders/pulsar-binder/spring-cloud-stream-binder-pulsar/src/test/java/org/springframework/cloud/stream/binder/pulsar/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)); @@ -95,7 +96,8 @@ public class PulsarExtendedBindingPropertiesTests { bind(props); assertThat(properties.getBindings()).containsOnlyKeys("my-foo"); - Map consumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties(); + Map consumerProps = PulsarBinderUtils + .convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo")); // Verify that the props can be loaded in a ConsumerBuilder assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, new ConsumerConfigurationData<>(), ConsumerConfigurationData.class)); @@ -121,7 +123,8 @@ public class PulsarExtendedBindingPropertiesTests { bind(props); - var bindingConsumerProps = properties.getExtendedConsumerProperties("my-foo").buildProperties(); + var bindingConsumerProps = PulsarBinderUtils + .convertConsumerPropertiesToMap(properties.getExtendedConsumerProperties("my-foo")); PulsarContainerProperties pulsarContainerProperties = new PulsarContainerProperties(); pulsarContainerProperties.getPulsarConsumerProperties().putAll(bindingConsumerProps); diff --git a/buildPulsarBinder.sh b/buildPulsarBinder.sh new file mode 100755 index 000000000..5bcce600a --- /dev/null +++ b/buildPulsarBinder.sh @@ -0,0 +1,14 @@ +#!/bin/bash + +if [ "$#" -eq 1 ]; then + ARGVALUE=$1 + if [[ $ARGVALUE == *"skipTest"* ]]; then + echo "Skippping Tests" + ./mvnw clean install -f binders/pulsar-binder/pom.xml -DskipTests + elif [[ $ARGVALUE == *"disable.checks"* ]]; then + echo "Skippping checkstyle checks" + ./mvnw clean install -f binders/pulsar-binder/pom.xml -Ddisable.checks=true + fi +else + ./mvnw clean install -f binders/pulsar-binder/pom.xml +fi