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 3740a9f6..7c16804f 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 @@ -27,12 +27,15 @@ import java.util.regex.Pattern; 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.MessageId; 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.common.schema.SchemaType; @@ -132,6 +135,16 @@ public class PulsarProperties { */ private SubscriptionType subscriptionType = SubscriptionType.Exclusive; + /** + * Map of properties to add to the subscription. + */ + private Map subscriptionProperties = new HashMap<>(); + + /** + * Subscription mode to be used when subscribing to the topic. + */ + private SubscriptionMode subscriptionMode = SubscriptionMode.Durable; + /** * Number of messages that can be accumulated before the consumer calls "receive". */ @@ -205,17 +218,60 @@ public class PulsarProperties { */ private RegexSubscriptionMode regexSubscriptionMode = RegexSubscriptionMode.PersistentOnly; + /** + * Dead letter policy to use. + */ + @Nullable + private DeadLetterPolicy deadLetterPolicy; + + /** + * Whether to auto retry messages. + */ + private Boolean retryEnable = false; + /** * Whether the consumer auto-subscribes for partition increase. This is only for * partitioned consumers. */ private Boolean autoUpdatePartitions = true; + /** + * Interval of partitions discovery updates. + */ + private Duration autoUpdatePartitionsInterval = Duration.ofMinutes(1); + /** * Whether to replicate subscription state. */ private Boolean replicateSubscriptionState = false; + /** + * Whether to include the given position of any reset operation like + * {@link org.apache.pulsar.client.api.Consumer#seek(long) or + * {@link Consumer#seek(MessageId)}}. + */ + private Boolean resetIncludeHead = false; + + /** + * Whether the batch index acknowledgment is enabled. + */ + private Boolean batchIndexAckEnabled = false; + + /** + * Whether an acknowledgement receipt is enabled. + */ + private Boolean ackReceiptEnabled = false; + + /** + * Whether pooling of messages and the underlying data buffers is enabled. + */ + private Boolean poolMessages = false; + + /** + * Whether to start the consumer in a paused state. + */ + private Boolean startPaused = false; + /** * Whether to automatically drop outstanding un-acked messages if the queue is * full. @@ -257,6 +313,22 @@ public class PulsarProperties { this.subscriptionName = subscriptionName; } + public Map getSubscriptionProperties() { + return this.subscriptionProperties; + } + + public void setSubscriptionProperties(Map subscriptionProperties) { + this.subscriptionProperties = subscriptionProperties; + } + + public SubscriptionMode getSubscriptionMode() { + return this.subscriptionMode; + } + + public void setSubscriptionMode(SubscriptionMode subscriptionMode) { + this.subscriptionMode = subscriptionMode; + } + public SubscriptionType getSubscriptionType() { return this.subscriptionType; } @@ -345,7 +417,7 @@ public class PulsarProperties { this.properties = properties; } - public Boolean isReadCompacted() { + public Boolean getReadCompacted() { return this.readCompacted; } @@ -377,7 +449,24 @@ public class PulsarProperties { this.regexSubscriptionMode = regexSubscriptionMode; } - public Boolean isAutoUpdatePartitions() { + @Nullable + public DeadLetterPolicy getDeadLetterPolicy() { + return this.deadLetterPolicy; + } + + public void setDeadLetterPolicy(@Nullable DeadLetterPolicy deadLetterPolicy) { + this.deadLetterPolicy = deadLetterPolicy; + } + + public Boolean getRetryEnable() { + return this.retryEnable; + } + + public void setRetryEnable(Boolean retryEnable) { + this.retryEnable = retryEnable; + } + + public Boolean getAutoUpdatePartitions() { return this.autoUpdatePartitions; } @@ -385,7 +474,15 @@ public class PulsarProperties { this.autoUpdatePartitions = autoUpdatePartitions; } - public Boolean isReplicateSubscriptionState() { + public Duration getAutoUpdatePartitionsInterval() { + return this.autoUpdatePartitionsInterval; + } + + public void setAutoUpdatePartitionsInterval(Duration autoUpdatePartitionsInterval) { + this.autoUpdatePartitionsInterval = autoUpdatePartitionsInterval; + } + + public Boolean getReplicateSubscriptionState() { return this.replicateSubscriptionState; } @@ -393,7 +490,47 @@ public class PulsarProperties { this.replicateSubscriptionState = replicateSubscriptionState; } - public Boolean isAutoAckOldestChunkedMessageOnQueueFull() { + public Boolean getResetIncludeHead() { + return this.resetIncludeHead; + } + + public void setResetIncludeHead(Boolean resetIncludeHead) { + this.resetIncludeHead = resetIncludeHead; + } + + public Boolean getBatchIndexAckEnabled() { + return this.batchIndexAckEnabled; + } + + public void setBatchIndexAckEnabled(Boolean batchIndexAckEnabled) { + this.batchIndexAckEnabled = batchIndexAckEnabled; + } + + public Boolean getAckReceiptEnabled() { + return this.ackReceiptEnabled; + } + + public void setAckReceiptEnabled(Boolean ackReceiptEnabled) { + this.ackReceiptEnabled = ackReceiptEnabled; + } + + public Boolean getPoolMessages() { + return this.poolMessages; + } + + public void setPoolMessages(Boolean poolMessages) { + this.poolMessages = poolMessages; + } + + public Boolean getStartPaused() { + return this.startPaused; + } + + public void setStartPaused(Boolean startPaused) { + this.startPaused = startPaused; + } + + public Boolean getAutoAckOldestChunkedMessageOnQueueFull() { return this.autoAckOldestChunkedMessageOnQueueFull; } @@ -426,6 +563,8 @@ public class PulsarProperties { map.from(this::getTopicsPattern).to(properties.in("topicsPattern")); map.from(this::getSubscriptionName).to(properties.in("subscriptionName")); map.from(this::getSubscriptionType).to(properties.in("subscriptionType")); + map.from(this::getSubscriptionProperties).to(properties.in("subscriptionProperties")); + map.from(this::getSubscriptionMode).to(properties.in("subscriptionMode")); map.from(this::getReceiverQueueSize).to(properties.in("receiverQueueSize")); map.from(this::getAcknowledgementsGroupTime).as(it -> it.toNanos() / 1000) .to(properties.in("acknowledgementsGroupTimeMicros")); @@ -439,13 +578,22 @@ public class PulsarProperties { map.from(this::getPriorityLevel).to(properties.in("priorityLevel")); map.from(this::getCryptoFailureAction).to(properties.in("cryptoFailureAction")); map.from(this::getProperties).to(properties.in("properties")); - map.from(this::isReadCompacted).to(properties.in("readCompacted")); + map.from(this::getReadCompacted).to(properties.in("readCompacted")); map.from(this::getSubscriptionInitialPosition).to(properties.in("subscriptionInitialPosition")); map.from(this::getPatternAutoDiscoveryPeriod).to(properties.in("patternAutoDiscoveryPeriod")); map.from(this::getRegexSubscriptionMode).to(properties.in("regexSubscriptionMode")); - map.from(this::isAutoUpdatePartitions).to(properties.in("autoUpdatePartitions")); - map.from(this::isReplicateSubscriptionState).to(properties.in("replicateSubscriptionState")); - map.from(this::isAutoAckOldestChunkedMessageOnQueueFull) + map.from(this::getDeadLetterPolicy).to(properties.in("deadLetterPolicy")); + map.from(this::getRetryEnable).to(properties.in("retryEnable")); + map.from(this::getAutoUpdatePartitions).to(properties.in("autoUpdatePartitions")); + map.from(this::getAutoUpdatePartitionsInterval).as(Duration::toSeconds) + .to(properties.in("autoUpdatePartitionsIntervalSeconds")); + map.from(this::getReplicateSubscriptionState).to(properties.in("replicateSubscriptionState")); + map.from(this::getResetIncludeHead).to(properties.in("resetIncludeHead")); + map.from(this::getBatchIndexAckEnabled).to(properties.in("batchIndexAckEnabled")); + map.from(this::getAckReceiptEnabled).to(properties.in("ackReceiptEnabled")); + map.from(this::getPoolMessages).to(properties.in("poolMessages")); + map.from(this::getStartPaused).to(properties.in("startPaused")); + map.from(this::getAutoAckOldestChunkedMessageOnQueueFull) .to(properties.in("autoAckOldestChunkedMessageOnQueueFull")); map.from(this::getMaxPendingChunkedMessage).to(properties.in("maxPendingChunkedMessage")); map.from(this::getExpireTimeOfIncompleteChunkedMessage).as(Duration::toMillis) 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 a977df0d..0ed56e45 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 @@ -18,6 +18,7 @@ package org.springframework.pulsar.autoconfigure; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.assertj.core.api.Assertions.assertThatNoException; import java.util.Collection; import java.util.Collections; @@ -27,14 +28,17 @@ import java.util.Set; 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.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; @@ -145,8 +149,8 @@ public class PulsarPropertiesTests { Map producerProps = properties.buildProducerProperties(); // Verify that the props can be loaded in a ProducerBuilder - ConfigurationDataUtils.loadData(producerProps, new ProducerConfigurationData(), - ProducerConfigurationData.class); + assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(producerProps, + new ProducerConfigurationData(), ProducerConfigurationData.class)); assertThat(producerProps).containsEntry("topicName", "my-topic") .containsEntry("producerName", "my-producer").containsEntry("sendTimeoutMs", 2_000) @@ -181,6 +185,8 @@ public class PulsarPropertiesTests { 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"); @@ -195,8 +201,19 @@ public class PulsarPropertiesTests { 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", "AllTopics"); + 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"); @@ -204,13 +221,20 @@ public class PulsarPropertiesTests { bind(props); Map consumerProps = properties.buildConsumerProperties(); + // Verify that the props can be loaded in a ConsumerBuilder + assertThatNoException().isThrownBy(() -> ConfigurationDataUtils.loadData(consumerProps, + new ConsumerConfigurationData<>(), ConsumerConfigurationData.class)); + assertThat(consumerProps) .hasEntrySatisfying("topicNames", n -> assertThat((Collection) n).containsExactly("my-topic")) .hasEntrySatisfying("topicsPattern", p -> assertThat(p.toString()).isEqualTo("my-pattern")) .containsEntry("subscriptionName", "my-subscription") - .containsEntry("subscriptionType", SubscriptionType.Shared).containsEntry("receiverQueueSize", 1) - .containsEntry("acknowledgementsGroupTimeMicros", 2_000_000L) + .containsEntry("subscriptionType", SubscriptionType.Shared) + .hasEntrySatisfying("subscriptionProperties", + p -> assertThat((Map) p).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) @@ -222,7 +246,17 @@ public class PulsarPropertiesTests { .containsEntry("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest) .containsEntry("patternAutoDiscoveryPeriod", 9) .containsEntry("regexSubscriptionMode", RegexSubscriptionMode.AllTopics) - .containsEntry("autoUpdatePartitions", false).containsEntry("replicateSubscriptionState", true) + .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/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java index b9221283..01acb824 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ConsumerBuilderConfigurationUtil.java @@ -60,6 +60,7 @@ public final class ConsumerBuilderConfigurationUtil { propertiesCopy.remove("cryptoKeyReader"); propertiesCopy.remove("messageCrypto"); propertiesCopy.remove("batchReceivePolicy"); + propertiesCopy.remove("keySharedPolicy"); propertiesCopy.remove("payloadProcessor"); builder.loadConf(propertiesCopy); @@ -79,6 +80,8 @@ public final class ConsumerBuilderConfigurationUtil { "messageCrypto"); applyValueToBuilderAfterLoadConf(builderConf::getBatchReceivePolicy, builder::batchReceivePolicy, properties, "batchReceivePolicy"); + applyValueToBuilderAfterLoadConf(builderConf::getKeySharedPolicy, builder::keySharedPolicy, properties, + "keySharedPolicy"); applyValueToBuilderAfterLoadConf(builderConf::getPayloadProcessor, builder::messagePayloadProcessor, properties, "payloadProcessor"); }