Past Acknowledged messages are redelivered

[Inbasasi]

Apache Pulsar Version - 2.11.1

While using Apache Pulsar delayed functionality, messages are produced with a delay.
We are facing issues recently that the past Acknowledged messages (14 days older and also the most recent acknowledged message) are being redelivered once again.

We also have configured a message retention policy.

{
  "retentionTimeInMinutes" : 360,
  "retentionSizeInMB" : 3072
}

Can someone help why past acknowledged messages are consumed again?

Pulsar Config

 public Consumer<Event> pulsarConsumer(PulsarClient pulsarClient,
                        PulsarProperties pulsarProperties)
                        throws PulsarClientException {
                return pulsarClient.newConsumer(Schema.JSON(¸.class)).ackTimeout(600000, TimeUnit.MILLISECONDS)
                                .consumerName("event-handler-1").topic(pulsarProperties.getTopic())
                                .subscriptionName("event-handler")
                                .subscriptionType(SubscriptionType.Shared)
                                .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(5).build())
                                .autoUpdatePartitions(true).receiverQueueSize(2000)
                                .subscriptionInitialPosition(pulsarProperties.getEventPosition())
                                .messageListener(consumer).subscribe();

[lhotari]

One possible reason is simply bugs in the Pulsar client or broker were acknowledgements were lost. If an acknowledgement is missing, messages would get redelivered. The redelivery moment depends on the subscription type and client settings. There have been bugs in the past where acknowledgements were lost due to client or broker bugs, however those are a rare minority of the problems.

There are cases where so called individual acknowledgements ("individually deleted", nonContiguousDeletedMessagesRanges in stats, totalNonContiguousDeletedMessagesRange in stats-internal) can overflow the limits.
By default, Pulsar will keep up to 10000 of such "ack holes". This is explained in the comment of managedLedgerMaxUnackedRangesToPersist:

pulsar/conf/broker.conf

Lines 1274 to 1281 in bcf8afb

	# Max number of "acknowledgment holes" that are going to be persistently stored.
	# When acknowledging out of order, a consumer will leave holes that are supposed
	# to be quickly filled by acking all the messages. The information of which
	# messages are acknowledged is persisted by compressing in "ranges" of messages
	# that were acknowledged. After the max number of ranges is reached, the information
	# will only be tracked in memory and messages will be redelivered in case of
	# crashes.
	managedLedgerMaxUnackedRangesToPersist=10000

When adjusting this limit, it's possible that the maximum size of a single Netty message or ZNode is reached.
The Netty message max size towards Bookkeeper is 5MB by default (and can be adjusted with nettyMaxFrameSizeByte setting in bookkeeper.conf) . The ZNode max size is 10MB by default and can be increased with tradeoffs. The Zookeeper system property is jute.maxbuffer. When changing this setting, it is necessary to ensure that it's changed across all Zookeeper nodes and Zookeeper clients, which are the brokers, bookies and other Pulsar components that might access Zookeeper.

To reduce the size of ZNodes, there's ManagedCursorInfo compression which was introduced in PIP-146. (ManagedLedgerInfo compression was introduced earlier, in PR #11490.)
ManagedCursorInfo compression can enable using a high managedLedgerMaxUnackedRangesToPersist setting.

For allowing higher managedLedgerMaxUnackedRangesToPersist setting, there's also a later improvement, PR #9292 which was recently also cherry-picked to branch-3.0 and will be available in the upcoming 3.0.10 release. The feature will be enabled when managedLedgerPersistIndividualAckAsLongArray is set to true.

In your case, you'd need to check your topic's stats (or partitioned-stats) to see how large the nonContiguousDeletedMessagesRanges value is. If it's more than 10000, you would be losing acknowledgement state.

There are also many improvements where the nonContiguousDeletedMessagesRanges is minimized. In Pulsar 4.0, for Key_Shared subscriptions there's #23231 which will significantly reduce the amount of "ack holes" with Key_Shared subscriptions.

There was also PIP-299: Stop dispatch messages if the individual acks will be lost in the persistent storage which introduced a setting dispatcherPauseOnAckStatePersistentEnabled. That solution helps reduce the overflow, but it won't completely prevent it. In testing, I have seen managedLedgerMaxUnackedRangesToPersist go over limits even when dispatcherPauseOnAckStatePersistentEnabled=true. The PIP-299 document contains more details about the managedLedgerMaxUnackedRangesToPersist overflow problem.

Since you are running on 2.11.1, you are way behind in sticking to a maintained and supported Pulsar version. In the OSS project, we would request Pulsar users to contribute in a way where they first try to reproduce the issue with a supported version since we don't maintain old versions. This is explained in the Pulsar release policy and supported versions document.

Newer Pulsar clients are compatible with all older Pulsar broker versions with some small exceptions. The latest Pulsar 3.0.x or 4.0.x client is compatible with Pulsar 2.11.x brokers. On your side, the first step would be to upgrade clients and then get to a maintained Pulsar version. In your case, you'd first have to upgrade to 2.11.4 and then to 3.0.9 to get on a supported Pulsar broker version. Upgrade compatibility between release versions is explained in the release policy document.

[Inbasasi]

Thanks, We will check the topic stats & possibly perform a upgrade
Also, the other issue that we face here is the messages are not cleared after 6 hours which is the retention policy.
We can see all the old messages still in the bookie.

Any idea why the storage is not cleared event after acknowledgment?
Whether this will not work for delayed messages?

[lhotari]

Any idea why the storage is not cleared event after acknowledgment?

In Pulsar, retention will by default only impact acknowledged messages, so delayed messages will prevent recovering storage space. Pulsar will delete a ledger from bookkeeper only after all messages in a single ledger have been acknowledged.

There's a possibility to use TTL to automatically acknowledge (or expire) messages by time.

There's more details in https://pulsar.apache.org/docs/4.0.x/cookbooks-retention-expiry/.

Whether this will not work for delayed messages?

Regarding retention and acknowledgements, delayed messages are like any other message which hasn't been acknowledged.

A large amount of delayed message will also add to the nonContiguousDeletedMessagesRanges when delayed messages are very sparse and some messages are delayed and some not. The previous message of increasing the limits with managedLedgerPersistIndividualAckAsLongArray=true (since upcoming 3.0.10 and 4.0.3) and bumping managedLedgerMaxUnackedRangesToPersist=500000. (you could also set managedCursorInfoCompressionType=LZ4, requires PULSAR_PREFIX_managedCursorInfoCompressionType: 'LZ4' since the value isn't in broker.conf and managedLedgerInfoCompressionType=LZ4/PULSAR_PREFIX_managedLedgerInfoCompressionType: 'LZ4').
You should also use delayedDeliveryTrackerFactoryClassName=org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory with a large amount of delayed messages to get the benefits of PIP-195: New bucket based delayed message tracker.

[Inbasasi]

Attached here are the topic stats for reference.
test2.txt

[Inbasasi]

"nonContiguousDeletedMessagesRanges" : 843989,

The is the count of nonContiguousDeletedMessagesRanges in our case, As this is greater than 10000 (default managedLedgerMaxUnackedRangesToPersist=10000).
So it is expected we may lose the track of acks when the broker restarts?

Can we increase the config managedLedgerMaxUnackedRangesToPersist more to accommodate this?

Since our primary use case is delayed delivery, we might see out-of-sequence acks for sure.

[lhotari]

"nonContiguousDeletedMessagesRanges" : 843989,

The is the count of nonContiguousDeletedMessagesRanges in our case, As this is greater than 10000 (default managedLedgerMaxUnackedRangesToPersist=10000). So it is expected we may lose the track of acks when the broker restarts?

Yes. Not only when a broker restarts, but when the topic (as part of a namespace bundle) gets moved to another broker by the Pulsar load balancer.

Can we increase the config managedLedgerMaxUnackedRangesToPersist more to accommodate this?

I have answered to this questions in the previous answers above. You might hit the storage limit of 5MB if you increase it too much unless you are using the configurations available in Pulsar 3.0.x and 4.0.x .

Since our primary use case is delayed delivery, we might see out-of-sequence acks for sure.

Yes. Please read carefully my previous answers about other considerations.

There's also an additional change that you could make on your application side. Instead of using a single topic for delayed messages, it's possible to use topics that are daily/weekly/monthly. You'd place the delayed message in the topic that matches the time base bucket. In consuming side, you could consume to these topics with a regex based subscription. The benefit of this approach is that storage space could potentially be released earlier than in a single topic solution since the messages will be consumed during the period that the topic is for. Besides reducing disk space, this also reduces the number of "ack holes" (nonContiguousDeletedMessagesRanges) and would also scale better if you have a very high number of delayed messages.
This type of solution obviously adds complexity to the applications producing delayed messages. However when using an internal library, this complexity could be encapsulated.

[lhotari]

In your stats, the nonContiguousDeletedMessagesRangesSerializedSize is way below 5MB, so as a quick mitigation, you might want to increase managedLedgerMaxUnackedRangesToPersist . Increasing it too high could cause the state to be lost completely, so that's the tradeoff. With high nonContiguousDeletedMessagesRanges, you'll also have a challenge any time a broker restarts or the topic gets moved since all messages will be read from the oldest delayed (unacknowledged message). That's why you should migrated to 3.0.x or 4.0.x so that you could use BucketDelayedDeliveryTrackerFactory which solves the problem. (the previous answers contain details).

[Inbasasi]

thanks for the clarifications, I will check & get back if anything required

[Inbasasi]

Hi @lhotari we increased the nonContiguousDeletedMessagesRanges in config, Now the nonContiguousDeletedMessagesRangesSerializedSize for the entire topic (of 6 partition) is 10MB

What issue we might face if it is greater than 5 MB?

[lhotari]

I don't think that the behavior is defined in that case. You would have to test what happens. Most likely at least the state would get lost, but it could also cause unavailability issues when storing of the state fails when it goes over 5 MB. I don't have all answer off the top of my head and if you need more immediate and detailed answers, it's useful to subscribe to a commercial Pulsar support offering.
The Apache Pulsar project release and support policy is described in https://pulsar.apache.org/contribute/release-policy/.

[Inbasasi]

Hi @lhotari , Can you please check this?

[lhotari]

please see #23990 (reply in thread)