Request support for configuring the number of threads for a single Spring-Pulsar consumer

[Programmer-yyds]

Currently, a single Spring-Pulsar consumer consumes messages using a single thread by default, which becomes a bottleneck in high-throughput scenarios.

Feature request:

Support configuring the number of threads for a single consumer to improve the processing capacity of an individual consumer.

Comparison: The native Pulsar Client can achieve multi-threaded consumption via asynchronous processing or ConsumerListener.

[onobc]

Hi @Programmer-yyds , how is your app consuming messages w/ Spring Pulsar? When you say

Currently, a single Spring-Pulsar consumer consumes messages using a single thread by default

are you referring to the Pulsar consumer that is used by the @PulsarListener when consuming messages?

[Programmer-yyds]

Proposal: Enable Multi-threaded Consumption Within a Single Pulsar Consumer in Spring Pulsar

Background

With the native Pulsar Client, a single Consumer can efficiently leverage its internal receiverQueue through two primary approaches:

• Using consumer.receiveAsync() with a thread pool to dispatch messages concurrently.
• Using a MessageListener combined with an application thread pool for parallel message handling.

This design allows a single Consumer to process prefetched messages concurrently, fully utilizing the receiverQueueSize.

In contrast, when using Spring Pulsar:

• The DefaultPulsarMessageListenerContainer currently operates with a single-threaded event loop, sequentially retrieving messages and invoking the annotated @PulsarListener method.
• As a result, messages in the receiverQueueSize are still processed serially by a single thread, which does not take advantage of the concurrency features provided by the Pulsar client.
• Configuring concurrency = N simply creates N separate Consumer instances, which introduces additional subscription overhead and does not enable concurrency within a single Consumer.

Problem Statement

This limitation makes it challenging to achieve high-throughput, low-latency consumption scenarios where a single Consumer is expected to:

1. Efficiently process messages already prefetched into its receiverQueue.
2. Avoid excessive subscription rebalances and connection overhead caused by large concurrency = N values.
3. Preserve single Consumer semantics, which are particularly critical in subscription modes such as Key_Shared and Exclusive.

Proposal

Introduce support for multi-threaded consumption within a single Consumer in Spring Pulsar.

Option 1: Extend ListenerContainer with a configurable TaskExecutor

• By default, processing remains single-threaded (backward compatible).
• If a user configures taskExecutor or sets threads = N, messages pulled from the receiverQueue will be dispatched to the thread pool for concurrent handling.
• Acknowledge (ack) and negative acknowledge (nack) logic must be executed safely within the thread pool.

Pseudocode example:

ExecutorService executor = Executors.newFixedThreadPool(8);

consumer.receiveAsync().thenAccept(msg -> {
    executor.submit(() -> {
        try {
            listener.onMessage(msg);
            consumer.acknowledge(msg);
        } catch (Exception e) {
            consumer.negativeAcknowledge(msg);
        }
    });
});

Option 2: Add a threads attribute to @PulsarListener

@PulsarListener(
    subscriptionName = "my-sub",
    topics = "my-topic",
    threads = 8 // Number of threads per Consumer instance
)
public void consume(String msg) {
    ...
}

Option 3: Compatibility Considerations

• concurrency > 1 → continues to create multiple Consumer instances (as it does today).
• threads > 1 → enables multi-threaded processing within a single Consumer.
• Both options can be combined to balance subscription scaling and intra-consumer concurrency.

Benefits

• Better utilization of prefetched messages (receiverQueueSize).
• Reduced subscription and connection overhead compared to high concurrency values.
• Maintains single-Consumer semantics while still enabling high throughput.
• Provides flexibility to users to balance between multiple Consumer scaling and intra-consumer concurrency.

@onobc could you please take a look and consider fixing this? 🙏
Thanks again!

[Programmer-yyds]

Description:

1️⃣ Native Pulsar Behavior

Partitioned Topic

• Suppose you have a topic my-topic with 4 partitions.
• When creating a consumer:

Consumer<String> consumer = client.newConsumer(Schema.STRING)
    .topic("persistent://public/pulsar/my-topic")
    .subscriptionName("sub")
    .subscriptionType(SubscriptionType.Shared)
    .subscribe();

• Actual behavior:
  • Pulsar creates a ConsumerImpl instance for each partition.
  • Each partition consumer has its own receiver queue.
  • Each consumer asynchronously receives messages using its own internal thread.

topicsPattern

• When subscribing with a regex pattern:

Consumer<String> consumer = client.newConsumer(Schema.STRING)
    .topicsPattern("persistent://public/pulsar/mys-.*")
    .subscriptionName("sub")
    .subscriptionType(SubscriptionType.Shared)
    .subscribe();

• Pulsar:
  • Matches all topics that fit the regex.
  • Creates an internal consumer for each matched topic.
  • Each internal consumer has its own thread to pull messages.

✅ Conclusion:
For both partitioned topics and topicsPattern, Pulsar internally creates separate consumer instances for each partition/topic, each with independent message reception.

---

2️⃣ Spring Pulsar Default Consumption Behavior

Using Spring Boot’s @PulsarListener:

@PulsarListener(
    subscriptionName = "my-sub",
    topics = "persistent://public/pulsar/my-topic",
    subscriptionType = SubscriptionType.Shared
)
public void consume(String msg) {
    System.out.println(Thread.currentThread().getName() + " received: " + msg);
}

• Spring invokes the listener in a single thread by default.
• Even if the underlying Pulsar client has multiple consumers per partition/topic, Spring Listener Container only uses one thread to call consume().

---

❗ Request / Enhancement

Introduce multi-threaded consumption per single Consumer instance in Spring Pulsar, especially for partitioned topics and topicsPattern subscriptions, to improve concurrency and resource utilization.

• Each Consumer instance can have multiple threads calling the listener concurrently.
• For partitioned topics or multi-topic subscriptions via regex, this can fully leverage the underlying consumer instances and improve throughput and concurrency.
• Alternatively, the number of listener threads could be automatically set based on the number of ConsumerImpl instances per partition, with the total number of threads greater than or equal to the number of underlying consumers.

@onobc could you please take a look and consider fixing this? 🙏
Thanks again!