Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add more settings to Producer in PulsarProperties (#222)

Browse Source
This commit is contained in:
Christophe Bornet
2022-11-19 06:07:36 +01:00
committed by GitHub
parent 27b09d3b49
commit 9baa08083e
2 changed files with 168 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

@@ -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<String> 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<String, String> 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<String> getEncryptionKeys() {
return this.encryptionKeys;
}
public void setEncryptionKeys(Set<String> 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<String, String> getProperties() {
return this.properties;
}
public void setProperties(Map<String, String> 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;
}

31
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java
View File

@@ -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<String, String> 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<String, Object> 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<String>) keys)).containsExactly("my-key"))
.containsEntry("compressionType", CompressionType.LZ4).containsEntry("initialSequenceId", 9L)
.containsEntry("accessMode", ProducerAccessMode.Exclusive)
.containsEntry("lazyStartPartitionedProducers", true)
.hasEntrySatisfying("properties", properties -> assertThat(((Map<String, String>) properties))
.containsEntry("my-prop", "my-prop-value"));
}
}