diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveProperties.java index 549c7825..e532348f 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveProperties.java @@ -17,7 +17,11 @@ package org.springframework.pulsar.autoconfigure; import java.time.Duration; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.SortedMap; import java.util.TreeMap; import java.util.regex.Pattern; @@ -48,6 +52,8 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.PropertyMapper; +import org.springframework.lang.Nullable; +import org.springframework.util.unit.DataSize; import reactor.core.scheduler.Schedulers; @@ -146,11 +152,18 @@ public class PulsarReactiveProperties { */ private Duration batchingMaxPublishDelay = Duration.ofMillis(1); + private Integer roundRobinRouterBatchingPartitionSwitchFrequency; + /** * Maximum number of messages to be batched. */ private Integer batchingMaxMessages = 1000; + /** + * Maximum number of bytes permitted in a batch. + */ + private DataSize batchingMaxBytes = DataSize.ofKilobytes(128); + /** * Whether to automatically batch messages. */ @@ -161,16 +174,53 @@ public class PulsarReactiveProperties { */ private Boolean chunkingEnabled = false; + /** + * Names of the public encryption keys to use when encrypting data. + */ + private Set encryptionKeys = new HashSet<>(); + /** * Message compression type. */ private CompressionType compressionType; + /** + * Baseline for the sequence ids for messages published by the producer. + */ + @Nullable + private Long initialSequenceId; + + /** + * Whether partitioned producer automatically discover new partitions at runtime. + */ + private Boolean autoUpdatePartitions = true; + + /** + * Interval of partitions discovery updates. + */ + private Duration autoUpdatePartitionsInterval = Duration.ofMinutes(1); + + /** + * Whether the multiple schema mode is enabled. + */ + private Boolean multiSchema = true; + /** * Type of access to the topic the producer requires. */ private ProducerAccessMode producerAccessMode = ProducerAccessMode.Shared; + /** + * Whether producers in Shared mode register and connect immediately to the owner + * broker of each partition or start lazily on demand. + */ + private Boolean lazyStartPartitionedProducers = false; + + /** + * Map of properties to add to the producer. + */ + private Map properties = new HashMap<>(); + private final Cache cache = new Cache(); public String getTopicName() { @@ -245,6 +295,15 @@ public class PulsarReactiveProperties { this.batchingMaxPublishDelay = batchingMaxPublishDelay; } + public Integer getRoundRobinRouterBatchingPartitionSwitchFrequency() { + return this.roundRobinRouterBatchingPartitionSwitchFrequency; + } + + public void setRoundRobinRouterBatchingPartitionSwitchFrequency( + Integer roundRobinRouterBatchingPartitionSwitchFrequency) { + this.roundRobinRouterBatchingPartitionSwitchFrequency = roundRobinRouterBatchingPartitionSwitchFrequency; + } + public Integer getBatchingMaxMessages() { return this.batchingMaxMessages; } @@ -253,6 +312,14 @@ public class PulsarReactiveProperties { this.batchingMaxMessages = batchingMaxMessages; } + public DataSize getBatchingMaxBytes() { + return this.batchingMaxBytes; + } + + public void setBatchingMaxBytes(DataSize batchingMaxBytes) { + this.batchingMaxBytes = batchingMaxBytes; + } + public Boolean getBatchingEnabled() { return this.batchingEnabled; } @@ -269,6 +336,14 @@ public class PulsarReactiveProperties { this.chunkingEnabled = chunkingEnabled; } + public Set getEncryptionKeys() { + return this.encryptionKeys; + } + + public void setEncryptionKeys(Set encryptionKeys) { + this.encryptionKeys = encryptionKeys; + } + public CompressionType getCompressionType() { return this.compressionType; } @@ -277,6 +352,39 @@ public class PulsarReactiveProperties { this.compressionType = compressionType; } + @Nullable + public Long getInitialSequenceId() { + return this.initialSequenceId; + } + + public void setInitialSequenceId(@Nullable Long initialSequenceId) { + this.initialSequenceId = initialSequenceId; + } + + public Boolean getAutoUpdatePartitions() { + return this.autoUpdatePartitions; + } + + public void setAutoUpdatePartitions(Boolean autoUpdatePartitions) { + this.autoUpdatePartitions = autoUpdatePartitions; + } + + public Duration getAutoUpdatePartitionsInterval() { + return this.autoUpdatePartitionsInterval; + } + + public void setAutoUpdatePartitionsInterval(Duration autoUpdatePartitionsInterval) { + this.autoUpdatePartitionsInterval = autoUpdatePartitionsInterval; + } + + public Boolean getMultiSchema() { + return this.multiSchema; + } + + public void setMultiSchema(Boolean multiSchema) { + this.multiSchema = multiSchema; + } + public ProducerAccessMode getProducerAccessMode() { return this.producerAccessMode; } @@ -285,6 +393,22 @@ public class PulsarReactiveProperties { this.producerAccessMode = producerAccessMode; } + public Boolean getLazyStartPartitionedProducers() { + return this.lazyStartPartitionedProducers; + } + + public void setLazyStartPartitionedProducers(Boolean lazyStartPartitionedProducers) { + this.lazyStartPartitionedProducers = lazyStartPartitionedProducers; + } + + public Map getProperties() { + return this.properties; + } + + public void setProperties(Map properties) { + this.properties = properties; + } + public Cache getCache() { return this.cache; } @@ -303,11 +427,21 @@ public class PulsarReactiveProperties { map.from(this::getHashingScheme).to(spec::setHashingScheme); map.from(this::getCryptoFailureAction).to(spec::setCryptoFailureAction); map.from(this::getBatchingMaxPublishDelay).to(spec::setBatchingMaxPublishDelay); + map.from(this::getRoundRobinRouterBatchingPartitionSwitchFrequency) + .to(spec::setRoundRobinRouterBatchingPartitionSwitchFrequency); map.from(this::getBatchingMaxMessages).to(spec::setBatchingMaxMessages); + map.from(this::getBatchingMaxBytes).asInt(DataSize::toBytes).to(spec::setBatchingMaxBytes); map.from(this::getBatchingEnabled).to(spec::setBatchingEnabled); map.from(this::getChunkingEnabled).to(spec::setChunkingEnabled); + map.from(this::getEncryptionKeys).to(spec::setEncryptionKeys); map.from(this::getCompressionType).to(spec::setCompressionType); + map.from(this::getInitialSequenceId).to(spec::setInitialSequenceId); + map.from(this::getAutoUpdatePartitions).to(spec::setAutoUpdatePartitions); + map.from(this::getAutoUpdatePartitionsInterval).to(spec::setAutoUpdatePartitionsInterval); + map.from(this::getMultiSchema).to(spec::setMultiSchema); map.from(this::getProducerAccessMode).to(spec::setAccessMode); + map.from(this::getLazyStartPartitionedProducers).to(spec::setLazyStartPartitionedProducers); + map.from(this::getProperties).to(spec::setProperties); return new ImmutableReactiveMessageSenderSpec(spec); } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactivePropertiesTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactivePropertiesTests.java index 0613f76d..c8c99fa2 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactivePropertiesTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarReactivePropertiesTests.java @@ -81,11 +81,17 @@ public class PulsarReactivePropertiesTests { props.put("spring.pulsar.reactive.sender.hashing-scheme", "Murmur3_32Hash"); props.put("spring.pulsar.reactive.sender.crypto-failure-action", "SEND"); props.put("spring.pulsar.reactive.sender.batching-max-publish-delay", "5s"); - props.put("spring.pulsar.reactive.sender.batching-max-messages", "6"); + props.put("spring.pulsar.reactive.sender.round-robin-router-batching-partition-switch-frequency", "6"); + props.put("spring.pulsar.reactive.sender.batching-max-messages", "7"); + props.put("spring.pulsar.reactive.sender.batching-max-bytes", "8"); props.put("spring.pulsar.reactive.sender.batching-enabled", "false"); props.put("spring.pulsar.reactive.sender.chunking-enabled", "true"); + props.put("spring.pulsar.reactive.sender.encryption-keys[0]", "my-key"); props.put("spring.pulsar.reactive.sender.compression-type", "LZ4"); + props.put("spring.pulsar.reactive.sender.initial-sequence-id", "9"); props.put("spring.pulsar.reactive.sender.producer-access-mode", "Exclusive"); + props.put("spring.pulsar.reactive.sender.lazy-start=partitioned-producers", "true"); + props.put("spring.pulsar.reactive.sender.properties[my-prop]", "my-prop-value"); bind(props); ReactiveMessageSenderSpec senderSpec = properties.buildReactiveMessageSenderSpec(); @@ -99,11 +105,17 @@ public class PulsarReactivePropertiesTests { assertThat(senderSpec.getHashingScheme()).isEqualTo(HashingScheme.Murmur3_32Hash); assertThat(senderSpec.getCryptoFailureAction()).isEqualTo(ProducerCryptoFailureAction.SEND); assertThat(senderSpec.getBatchingMaxPublishDelay()).isEqualTo(Duration.ofSeconds(5)); - assertThat(senderSpec.getBatchingMaxMessages()).isEqualTo(6); + assertThat(senderSpec.getRoundRobinRouterBatchingPartitionSwitchFrequency()).isEqualTo(6); + assertThat(senderSpec.getBatchingMaxMessages()).isEqualTo(7); + assertThat(senderSpec.getBatchingMaxBytes()).isEqualTo(8); assertThat(senderSpec.getBatchingEnabled()).isEqualTo(false); assertThat(senderSpec.getChunkingEnabled()).isEqualTo(true); + assertThat(senderSpec.getEncryptionKeys()).containsExactly("my-key"); assertThat(senderSpec.getCompressionType()).isEqualTo(CompressionType.LZ4); + assertThat(senderSpec.getInitialSequenceId()).isEqualTo(9); assertThat(senderSpec.getAccessMode()).isEqualTo(ProducerAccessMode.Exclusive); + assertThat(senderSpec.getLazyStartPartitionedProducers()).isTrue(); + assertThat(senderSpec.getProperties()).hasSize(1).containsEntry("my-prop", "my-prop-value"); } }