Implementing Shared Subscriptions in Apache Pulsar for Parallel Message Processing

The Scaling Bottleneck in Message Consumption

When a single consumer cannot keep pace with the volume of messages arriving on a Pulsar topic, the backlog grows, increasing end-to-end latency. While Exclusive and Failover subscription modes ensure strict ordering, they limit processing to a single active consumer. To scale throughput horizontally, you must distribute the message load across a pool of consumers.

Prerequisites

• A functional Apache Pulsar cluster (ZooKeeper, BookKeeper, and Broker services running).
• The Pulsar Java Client library added to your project dependencies.
• A topic already created or configured for automatic creation.

Configuring a Shared Consumer

To implement a shared subscription, you must explicitly set the SubscriptionType to Shared during the consumer creation process. If you do not specify a type, Pulsar defaults to Exclusive mode, which will reject any second consumer attempting to join the same subscription.

// Initialize the Pulsar Client
PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://localhost:6650")
    .build();

// Create a Shared Consumer
Consumer consumer = client.newConsumer()
    .topic("persistent://public/default/my-shared-topic")
    .subscriptionName("my-shared-subscription")
    .subscriptionType(SubscriptionType.Shared) // Required for load balancing
    .subscribe();

The Processing Loop and Acknowledgment

In Shared mode, the broker tracks acknowledgments for each individual message rather than a cumulative cursor. This ensures that if one consumer fails, only the messages it was currently processing are redelivered.

while (true) {
    Message msg = consumer.receive();
    try {
        // Process the message logic here
        System.out.println("Consumer " + consumer.hashCode() + " processed: " + new String(msg.getData()));
        
        // Mark message as successfully processed
        consumer.acknowledge(msg);
    } catch (Exception e) {
        // Trigger redelivery after a configured delay
        consumer.negativeAcknowledge(msg);
    }
}

Operational Comparison: Shared vs. Key_Shared

Choosing between Shared and Key_Shared depends on whether your business logic requires ordering for specific entities (e.g., all events for User ID 123 must be processed in sequence).

Feature	Shared	Key_Shared
Distribution	Round-robin	Hash of the message key
Ordering	No guarantee	Ordered per key
Scaling	High (any consumer can take any message)	High (keys are mapped to consumers)
Broker Overhead	Moderate (individual ACKs)	Higher (key-to-consumer mapping)

Verifying Load Distribution

To verify that the broker is correctly balancing the load, deploy three separate instances of your consumer application using the same subscriptionName. Produce a burst of messages to the topic and monitor the logs of each instance to ensure messages are interleaved.

To check the backlog and consumer connectivity via the CLI, run the following command on the Pulsar broker or a machine with pulsar-admin installed:

# Run as a user with permissions to access the Pulsar admin API
pulsar-admin topics stats persistent://public/default/my-shared-topic

Expected Result: The output should show multiple active consumers under the my-shared-subscription entry, and the msgBacklog count should decrease as all consumers process messages simultaneously.

Limitations and Risks

• Broker Memory: Because the broker must track individual acknowledgments for every unacknowledged message in a shared subscription, a very high number of unacknowledged messages can increase broker memory pressure.
• Ordering: If your application relies on the sequence of events, SubscriptionType.Shared will cause race conditions. Use Key_Shared instead.

Recovery and State Changes

Because this operation changes the subscription type on the broker, you may need to reset the subscription if you accidentally created it as Exclusive.

1. Stop all consumers attached to the subscription.
2. Delete the subscription using the admin CLI to clear the state: pulsar-admin subscriptions delete persistent://public/default/my-shared-topic/my-shared-subscription
3. Restart the consumers with the correct SubscriptionType.Shared configuration.