GH-3147 Add consumer priority support to Rabbit binder properties

Signed-off-by: belljun3395 <[email protected]>

Author: belljun3395

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -210,6 +210,16 @@ public enum StandardHeaders {
 	 */
 	private boolean reactiveAtMostOnce;
 
+	/**
+	 * Consumer priority level. NOTE: Kafka does not natively support consumer priority.
+	 * This property is provided for consistency across binders but has no effect in Kafka.
+	 * For even message distribution across servers, use partition assignment strategies
+	 * or create separate bindings with concurrency=1.
+	 * Default: -1 (not supported)
+	 * @since 4.2
+	 */
+	private int consumerPriority = -1;
+
 	/**
 	 * @return Container's ack mode.
 	 */
@@ -486,4 +496,12 @@ public void setReactiveAtMostOnce(boolean reactiveAtMostOnce) {
 		this.reactiveAtMostOnce = reactiveAtMostOnce;
 	}
 
+	public int getConsumerPriority() {
+		return this.consumerPriority;
+	}
+
+	public void setConsumerPriority(int consumerPriority) {
+		this.consumerPriority = consumerPriority;
+	}
+
 }

binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java

@@ -145,6 +145,19 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
 	 */
 	private boolean superStream;
 
+	/**
+	 * Consumer priority for this consumer. Higher values indicate higher priority.
+	 * Requires the queue to be declared with x-max-priority argument.
+	 * Valid range: 0-255. Default: -1 (no priority set).
+	 */
+	private int consumerPriority = -1;
+
+	/**
+	 * Maximum priority for the queue. When set, the queue will be declared with
+	 * x-max-priority argument. Valid range: 1-255. Default: -1 (not set).
+	 */
+	private int queueMaxPriority = -1;
+
 	public boolean isTransacted() {
 		return transacted;
 	}
@@ -325,6 +338,22 @@ public void setSuperStream(boolean superStream) {
 		this.superStream = superStream;
 	}
 
+	public int getConsumerPriority() {
+		return this.consumerPriority;
+	}
+
+	public void setConsumerPriority(int consumerPriority) {
+		this.consumerPriority = consumerPriority;
+	}
+
+	public int getQueueMaxPriority() {
+		return this.queueMaxPriority;
+	}
+
+	public void setQueueMaxPriority(int queueMaxPriority) {
+		this.queueMaxPriority = queueMaxPriority;
+	}
+
 	/**
 	 * Container type.
 	 * @author Gary Russell

binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java

@@ -608,6 +608,15 @@ private void additionalArgs(Map<String, Object> args, RabbitCommonProperties pro
 				: properties.getMaxLengthBytes();
 		Integer maxPriority = isDlq ? properties.getDlqMaxPriority()
 				: properties.getMaxPriority();
+
+		// Add queue max priority for consumer priority support
+		if (!isDlq && properties instanceof RabbitConsumerProperties) {
+			RabbitConsumerProperties consumerProps = (RabbitConsumerProperties) properties;
+			if (consumerProps.getQueueMaxPriority() > 0) {
+				maxPriority = consumerProps.getQueueMaxPriority();
+			}
+		}
+
 		Integer ttl = isDlq ? properties.getDlqTtl() : properties.getTtl();
 		boolean lazy = isDlq ? properties.isDlqLazy() : properties.isLazy();
 		String overflow = isDlq ? properties.getDlqOverflowBehavior()

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -21,6 +21,7 @@
 import java.io.StringWriter;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
@@ -604,6 +605,12 @@ else if (getApplicationContext() != null) {
 					q -> extension.getConsumerTagPrefix() + "#"
 							+ index.getAndIncrement());
 		}
+		// Set consumer priority if configured
+		if (extension.getConsumerPriority() >= 0) {
+			Map<String, Object> consumerArgs = new HashMap<>();
+			consumerArgs.put("x-priority", extension.getConsumerPriority());
+			listenerContainer.setConsumerArguments(consumerArgs);
+		}
 		listenerContainer.setApplicationContext(getApplicationContext());
 		return listenerContainer;
 	}

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -2732,6 +2732,33 @@ public void setField(String field) {
 		}
 
 	}
+
+	@Test
+	void testConsumerPriority() throws Exception {
+		RabbitTestBinder binder = getBinder();
+		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
+		consumerProperties.getExtension().setConsumerPriority(10);
+		consumerProperties.getExtension().setQueueMaxPriority(10);
+
+		DirectChannel moduleInputChannel = createBindableChannel("input", new BindingProperties());
+
+		Binding<MessageChannel> consumerBinding = binder.bindConsumer("priority.test", "priorityGroup",
+				moduleInputChannel, consumerProperties);
+
+		AbstractMessageListenerContainer container =
+				TestUtils.getPropertyValue(consumerBinding, "lifecycle.messageListenerContainer",
+						AbstractMessageListenerContainer.class);
+
+		Map<String, Object> consumerArgs = container.getConsumerArguments();
+
+		assertThat(consumerArgs).isNotNull();
+		assertThat(consumerArgs.get("x-priority")).isEqualTo(10);
+		assertThat(consumerProperties.getExtension().getConsumerPriority()).isEqualTo(10);
+		assertThat(consumerProperties.getExtension().getQueueMaxPriority()).isEqualTo(10);
+
+		consumerBinding.unbind();
+	}
+
 	// @checkstyle:on
 
 	public static class TestBatchingStrategy implements BatchingStrategy {