From 9baa08083e2859b4b6d027a85419d1b41460479a Mon Sep 17 00:00:00 2001 From: Christophe Bornet Date: Sat, 19 Nov 2022 06:07:36 +0100 Subject: [PATCH] Add more settings to Producer in PulsarProperties (#222) --- .../autoconfigure/PulsarProperties.java | 149 +++++++++++++++++- .../autoconfigure/PulsarPropertiesTests.java | 31 +++- 2 files changed, 168 insertions(+), 12 deletions(-) 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 8f5adc32..3740a9f6 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 @@ -18,6 +18,7 @@ package org.springframework.pulsar.autoconfigure; import java.time.Duration; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; import java.util.Set; import java.util.SortedMap; @@ -37,6 +38,7 @@ import org.apache.pulsar.common.schema.SchemaType; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.PropertyMapper; +import org.springframework.lang.Nullable; import org.springframework.pulsar.listener.AckMode; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -507,11 +509,22 @@ public class PulsarProperties { */ private Duration batchingMaxPublishDelay = Duration.ofMillis(1); + /** + * Partition switch frequency while batching of messages is enabled and using + * round-robin routing mode for non-keyed message. + */ + private Integer batchingPartitionSwitchFrequencyByPublishDelay = 10; + /** * 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. */ @@ -522,16 +535,53 @@ public class PulsarProperties { */ 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 Cache cache = new Cache(); public String getTopicName() { @@ -558,7 +608,7 @@ public class PulsarProperties { this.sendTimeout = sendTimeout; } - public Boolean isBlockIfQueueFull() { + public Boolean getBlockIfQueueFull() { return this.blockIfQueueFull; } @@ -614,6 +664,15 @@ public class PulsarProperties { this.batchingMaxPublishDelay = batchingMaxPublishDelay; } + public Integer getBatchingPartitionSwitchFrequencyByPublishDelay() { + return this.batchingPartitionSwitchFrequencyByPublishDelay; + } + + public void setBatchingPartitionSwitchFrequencyByPublishDelay( + Integer batchingPartitionSwitchFrequencyByPublishDelay) { + this.batchingPartitionSwitchFrequencyByPublishDelay = batchingPartitionSwitchFrequencyByPublishDelay; + } + public Integer getBatchingMaxMessages() { return this.batchingMaxMessages; } @@ -622,7 +681,15 @@ public class PulsarProperties { this.batchingMaxMessages = batchingMaxMessages; } - public Boolean isBatchingEnabled() { + public DataSize getBatchingMaxBytes() { + return this.batchingMaxBytes; + } + + public void setBatchingMaxBytes(DataSize batchingMaxBytes) { + this.batchingMaxBytes = batchingMaxBytes; + } + + public Boolean getBatchingEnabled() { return this.batchingEnabled; } @@ -630,7 +697,7 @@ public class PulsarProperties { this.batchingEnabled = batchingEnabled; } - public Boolean isChunkingEnabled() { + public Boolean getChunkingEnabled() { return this.chunkingEnabled; } @@ -638,6 +705,14 @@ public class PulsarProperties { this.chunkingEnabled = chunkingEnabled; } + public Set getEncryptionKeys() { + return this.encryptionKeys; + } + + public void setEncryptionKeys(Set encryptionKeys) { + this.encryptionKeys = encryptionKeys; + } + public CompressionType getCompressionType() { return this.compressionType; } @@ -646,6 +721,39 @@ public class PulsarProperties { 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; } @@ -654,6 +762,22 @@ public class PulsarProperties { 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; } @@ -665,8 +789,8 @@ public class PulsarProperties { map.from(this::getTopicName).to(properties.in("topicName")); map.from(this::getProducerName).to(properties.in("producerName")); - map.from(this::getSendTimeout).as(Duration::toMillis).to(properties.in("sendTimeoutMs")); - map.from(this::isBlockIfQueueFull).to(properties.in("blockIfQueueFull")); + 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")); @@ -675,11 +799,22 @@ public class PulsarProperties { 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::isBatchingEnabled).to(properties.in("batchingEnabled")); - map.from(this::isChunkingEnabled).to(properties.in("chunkingEnabled")); + 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; } 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 66131d7a..a977df0d 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 @@ -23,6 +23,7 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Set; import org.apache.pulsar.client.api.CompressionType; import org.apache.pulsar.client.api.ConsumerCryptoFailureAction; @@ -33,6 +34,8 @@ 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.SubscriptionType; +import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils; +import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; @@ -113,6 +116,7 @@ public class PulsarPropertiesTests { class ProducerPropertiesTests { @Test + @SuppressWarnings("unchecked") void producerProperties() { Map props = new HashMap<>(); props.put("spring.pulsar.producer.topic-name", "my-topic"); @@ -125,26 +129,43 @@ public class PulsarPropertiesTests { 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-max-messages", "6"); + 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 = properties.buildProducerProperties(); + // Verify that the props can be loaded in a ProducerBuilder + ConfigurationDataUtils.loadData(producerProps, new ProducerConfigurationData(), + ProducerConfigurationData.class); + assertThat(producerProps).containsEntry("topicName", "my-topic") - .containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000L) + .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("batchingMaxMessages", 6) + .containsEntry("batchingMaxPublishDelayMicros", 5_000_000L) + .containsEntry("batchingPartitionSwitchFrequencyByPublishDelay", 6) + .containsEntry("batchingMaxMessages", 7).containsEntry("batchingMaxBytes", 8) .containsEntry("batchingEnabled", false).containsEntry("chunkingEnabled", true) - .containsEntry("compressionType", CompressionType.LZ4) - .containsEntry("accessMode", ProducerAccessMode.Exclusive); + .hasEntrySatisfying("encryptionKeys", + keys -> assertThat(((Set) keys)).containsExactly("my-key")) + .containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L) + .containsEntry("accessMode", ProducerAccessMode.Exclusive) + .containsEntry("lazyStartPartitionedProducers", true) + .hasEntrySatisfying("properties", properties -> assertThat(((Map) properties)) + .containsEntry("my-prop", "my-prop-value")); } }