Problem Statement

Current Issue

Scenario: 2 servers × concurrency 2 = 4 consumers
Problem: Messages concentrate on one server → 200% CPU usage, other server idle

Root Cause: Messaging systems distribute messages in round-robin fashion to individual consumers (not servers), causing load concentration on specific servers during compute-intensive message processing.

Solution

Implemented Consumer Priority mechanism:

• Assign priority levels to consumers
• Higher priority consumers receive messages first
• Lower priority consumers process messages when high-priority ones are busy

Implementation Details

1. RabbitMQ Binder (Full Implementation)

New Properties

// RabbitConsumerProperties.java
private int consumerPriority = -1;  // Consumer priority (0-255)
private int queueMaxPriority = -1;  // Queue max priority (1-255)

Queue Configuration

// RabbitExchangeQueueProvisioner.java:613-618
if (!isDlq && properties instanceof RabbitConsumerProperties) {
    RabbitConsumerProperties consumerProps = (RabbitConsumerProperties) properties;
    if (consumerProps.getQueueMaxPriority() > 0) {
        maxPriority = consumerProps.getQueueMaxPriority();
    }
}
// Sets x-max-priority argument on queue

Consumer Configuration

// RabbitMessageChannelBinder.java:606-611
if (extension.getConsumerPriority() >= 0) {
    Map<String, Object> consumerArgs = new HashMap<>();
    consumerArgs.put("x-priority", extension.getConsumerPriority());
    listenerContainer.setConsumerArguments(consumerArgs);
}

Usage Example

spring:
  cloud:
    stream:
      rabbit:
        bindings:
          input-in-0:
            consumer:
              consumer-priority: 10      # Consumer priority
              queue-max-priority: 10     # Queue max priority

2. Pulsar Binder (Native Support)

Pulsar already supports priorityLevel in parent class PulsarProperties.Consumer, automatically mapped in toBaseConsumerPropertiesMap().

spring:
  cloud:
    stream:
      pulsar:
        bindings:
          input-in-0:
            consumer:
              priority-level: 5  # Range: 0-10

3. Kafka Binder (API Consistency)

Kafka does not natively support consumer priority. Property added for API consistency with clear documentation of non-support:

// KafkaConsumerProperties.java:213-221
/**
 * 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)
 */
private int consumerPriority = -1;

Usage Scenario

Before (Problem)

Server 1: Consumer1(busy), Consumer2(busy) → CPU 200%
Server 2: Consumer3(idle), Consumer4(idle) → CPU 0%

After (With Consumer Priority)

# Server 1
spring.cloud.stream.rabbit.bindings.input.consumer.consumer-priority: 10

# Server 2
spring.cloud.stream.rabbit.bindings.input.consumer.consumer-priority: 5

Server 1: Consumer1(priority), Consumer2(priority) → CPU 100%
Server 2: Consumer3(backup), Consumer4(backup) → CPU 100% (when Server 1 is busy)
→ Even load distribution ✅

Issue #3122 Resolution

Among the 3 proposed solutions in the issue:

1. "consumer priority to thread index" - Implemented for RabbitMQ and Pulsar
2. "Dynamically create consumers" - Requires separate implementation
3. "Multiple bindings with concurrency=1" - Documented as Kafka alternative