Deep Dive into Apache Pulsar Message Acknowledgment Mechanism

In Apache Pulsar, to avoid duplicate message delivery, consumers must acknowledge messages. When a message is consumed, the consumer sends an acknowledgment (ack) request to the Broker, which then marks the message as truly consumed. This article introduces Pulsar's message acknowledgment modes and how normal message acknowledgment is implemented on the Broker side.

Message Acknowledgment Modes:

Pulsar supports multiple subscription types (shared, key_shared, failover) and corresponding acknowledgment behaviors. The main acknowledgment modes include:

1. Single Message Acknowledgment (acknowledge) : Unlike other messaging systems, Pulsar allows multiple consumers to consume a single partition. Each message can be individually acknowledged. For example, if Consumer-B acks message 4 while messages 1-3 are with Consumer-A and messages 5-6 are unacknowledged, only messages 1,2,3,5,6 will be redelivered.

2. Cumulative Message Acknowledgment (acknowledgeCumulative) : Consumers can acknowledge a message and all previous messages by calling acknowledgeCumulative with a specific message ID, improving ack performance. This is not supported in shared or key_shared modes.

3. Individual Batch Message Acknowledgment : Requires enabling acknowledgmentAtBatchIndexLevelEnabled configuration. Allows acknowledging specific messages within a batch rather than the entire batch.

4. Negative Acknowledgment (negativeAcknowledge) : Clients send a redeliverUnacknowledgedMessages command to the Broker to request message redelivery.

Implementation Details:

The acknowledge and acknowledgeCumulative interfaces delegate to AcknowledgmentsGroupingTracker. For persistent subscriptions, PersistentAcknowledgmentsGroupingTracker is used. To ensure acknowledgment performance and avoid high-concurrency ack requests, Pulsar supports batch acknowledgments by default. Individual ack requests are stored in pendingIndividualAcks Set and sent in batches every 100ms or when exceeding 1000 pending acks.

For batch message individual acknowledgments, a Map (pendingIndividualBatchIndexAcks) is used with BitSet to track which messages in a batch are acknowledged. BitSet significantly reduces memory usage—1KB can record acknowledgment status for 8,192 messages.

Cumulative ack is simpler: the Tracker only stores the latest acknowledgment position (LedgerId and EntryId).

Negative Acknowledgment Implementation:

Negative acks are handled by NegativeAcksTracker, which uses PulsarClient's timing wheel (default ~33ms per tick) with a default delay of 1 minute. Messages are redelivered after the delay expires. This approach merges requests and allows global delay configuration without specifying time for each message.

Handling Unacknowledged Messages:

When consumers receive messages without acknowledgment, two scenarios occur: 1) If receive() was called or async callback is waiting, messages are tracked in unAckedMessageTracker using a time-wheel mechanism. When timeout occurs, messages are automatically redelivered via redeliverUnacknowledgedMessages command. 2) If messages are pre-fetched but receive() hasn't been called, they accumulate in the local queue. The Broker records these as pendingAck and won't redeliver unless the Consumer is closed.

The Broker maintains a RedeliveryTracker (InMemoryRedeliveryTracker) to track how many times each message has been redelivered.