diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 2518a9bb9..32e3ce452 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -236,6 +236,18 @@ dlqOverflowBehavior:: Action to take when `dlqMaxLength` or `dlqMaxLengthBytes` is exceeded; currently `drop-head` or `reject-publish` but refer to the RabbitMQ documentation. + Default: `none` +dlqQuorum.deliveryLimit:: +When `quorum.enabled=true`, set a delivery limit after which the message is dropped or dead-lettered. ++ +Default: none - broker default will apply. +dlqQuorum.enabled:: +When true, create a quorum dead letter queue instead of a classic queue. ++ +Default: false +dlqQuorum.initialQuorumSize:: +When `quorum.enabled=true`, set the initial quorum size. ++ +Default: none - broker default will apply. dlqTtl:: Default time to live to apply to the dead letter queue when declared (in milliseconds). + @@ -342,6 +354,18 @@ Otherwise the queue name is `destination.group`. This is useful, for example, when using Spring Cloud Stream to consume from an existing RabbitMQ queue. + Default: false. +quorum.deliveryLimit:: +When `quorum.enabled=true`, set a delivery limit after which the message is dropped or dead-lettered. ++ +Default: none - broker default will apply. +quorum.enabled:: +When true, create a quorum queue instead of a classic queue. ++ +Default: false +quorum.initialQuorumSize:: +When `quorum.enabled=true`, set the initial quorum size. ++ +Default: none - broker default will apply. recoveryInterval:: The interval between connection recovery attempts, in milliseconds. + @@ -542,6 +566,21 @@ Maximum priority of messages in the dead letter queue (0-255) Applies only when `requiredGroups` are provided and then only to those groups. + Default: `none` +dlqQuorum.deliveryLimit:: +When `quorum.enabled=true`, set a delivery limit after which the message is dropped or dead-lettered. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: none - broker default will apply. +dlqQuorum.enabled:: +When true, create a quorum dead letter queue instead of a classic queue. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: false +dlqQuorum.initialQuorumSize:: +When `quorum.enabled=true`, set the initial quorum size. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: none - broker default will apply. dlqTtl:: Default time (in milliseconds) to live to apply to the dead letter queue when declared. Applies only when `requiredGroups` are provided and then only to those groups. @@ -607,6 +646,21 @@ This is useful, for example, when using Spring Cloud Stream to consume from an e Applies only when `requiredGroups` are provided and then only to those groups. + Default: false. +quorum.deliveryLimit:: +When `quorum.enabled=true`, set a delivery limit after which the message is dropped or dead-lettered. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: none - broker default will apply. +quorum.enabled:: +When true, create a quorum queue instead of a classic queue. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: false +quorum.initialQuorumSize:: +When `quorum.enabled=true`, set the initial quorum size. +Applies only when `requiredGroups` are provided and then only to those groups. ++ +Default: none - broker default will apply. routingKeyExpression:: A SpEL expression to determine the routing key to use when publishing messages. For a fixed routing key, use a literal expression, such as `routingKeyExpression='my.routingKey'` in a properties file or `routingKeyExpression: '''my.routingKey'''` in a YAML file. diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java index 91874014e..318f3f646 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java @@ -211,6 +211,16 @@ public abstract class RabbitCommonProperties { */ private Map dlqBindingArguments = new HashMap<>(); + /** + * Configure the queue to be type quorum instead of classic. + */ + private QuorumConfig quorum = new QuorumConfig(); + + /** + * Configure the DLQ to be type quorum instead of classic. + */ + private QuorumConfig dlqQuorum = new QuorumConfig(); + public String getExchangeType() { return this.exchangeType; } @@ -484,4 +494,54 @@ public abstract class RabbitCommonProperties { this.dlqBindingArguments = dlqBindingArguments; } + public QuorumConfig getQuorum() { + return this.quorum; + } + + public void setQuorum(QuorumConfig quorum) { + this.quorum = quorum; + } + + public QuorumConfig getDlqQuorum() { + return this.dlqQuorum; + } + + public void setDlqQuorum(QuorumConfig dlqQuorum) { + this.dlqQuorum = dlqQuorum; + } + + public static class QuorumConfig { + + private boolean enabled; + + private Integer initialGroupSize; + + private Integer deliveryLimit; + + public boolean isEnabled() { + return this.enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public Integer getInitialGroupSize() { + return this.initialGroupSize; + } + + public void setInitialGroupSize(Integer initialGroupSize) { + this.initialGroupSize = initialGroupSize; + } + + public Integer getDeliveryLimit() { + return this.deliveryLimit; + } + + public void setDeliveryLimit(Integer deliveryLimit) { + this.deliveryLimit = deliveryLimit; + } + + } + } diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index 9e1c0ad8c..543ac8e5d 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -44,6 +44,7 @@ import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties.QuorumConfig; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.provisioning.ConsumerDestination; @@ -499,6 +500,7 @@ public class RabbitExchangeQueueProvisioner boolean lazy = isDlq ? properties.isDlqLazy() : properties.isLazy(); String overflow = isDlq ? properties.getDlqOverflowBehavior() : properties.getOverflowBehavior(); + QuorumConfig quorum = isDlq ? properties.getDlqQuorum() : properties.getQuorum(); if (expires != null) { args.put("x-expires", expires); } @@ -520,6 +522,15 @@ public class RabbitExchangeQueueProvisioner if (StringUtils.hasText(overflow)) { args.put("x-overflow", overflow); } + if (quorum != null && quorum.isEnabled()) { + args.put("x-queue-type", "quorum"); + if (quorum.getDeliveryLimit() != null) { + args.put("x-delivery-limit", quorum.getDeliveryLimit()); + } + if (quorum.getInitialGroupSize() != null) { + args.put("x-quorum-initial-group-size", quorum.getInitialGroupSize()); + } + } } public static String applyPrefix(String prefix, String name) { diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index bac722127..ef0d46a0a 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -82,6 +82,7 @@ import org.springframework.cloud.stream.binder.PartitionTestSupport; import org.springframework.cloud.stream.binder.PollableSource; import org.springframework.cloud.stream.binder.RequeueCurrentMessageException; import org.springframework.cloud.stream.binder.Spy; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties.QuorumConfig; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; @@ -399,6 +400,10 @@ public class RabbitBinderTests extends properties.getExtension().setPrefetch(20); properties.getExtension().setHeaderPatterns(new String[] { "foo" }); properties.getExtension().setTxSize(10); + QuorumConfig quorum = properties.getExtension().getQuorum(); + quorum.setEnabled(true); + quorum.setDeliveryLimit(10); + quorum.setInitialGroupSize(1); properties.setInstanceIndex(0); consumerBinding = binder.bindConsumer("props.0", "test", createBindableChannel("input", new BindingProperties()), properties);