[Bug] big individuallyDeletedMessages causes message dispatching hangs

[YanshuoH]

Search before reporting

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

User environment

• broker version: 4.0.8
• broker os: Linux pulsar-broker-1a-0 6.12.40-64.114.amzn2023.aarch64 javascript client #1 SMP Tue Aug 26 05:25:54 UTC 2025 aarch64 aarch64 aarch64 GNU/Linux
• java: openjdk version "17.0.12" 2024-07-16
• client: golang
• client version: 0.17.0
• client os: same as broker
• client java version: NaN

Issue Description

One of our scenario is check user's payment with variadic delay, from 10s to 1h indifferent.
My observation is that when the individuallyDeletedMessages becomes quite big (100,000+, and the setting managedLedgerMaxUnackedRangesToPersist is 100,000 too), dispatching of messages become strange. The message dispatch is very slow and most messages don't get dispatched.
Checking the internal-stats, I can see something as such:

      "numberOfEntriesSinceFirstNotAckedMessage": 751170,
      "totalNonContiguousDeletedMessagesRange": 105911,

No more error message on both client and server side.

I see there's a similar issue #23200, yet we're using Shared subscription type.

Error messages

The suspicious message I got is:
client side tries to reconnect to the broker with:

INFO[0960] Connecting to broker                          remote_addr="pulsar://pulsar-broker.pulsar1.svc.cluster.local:6650"
INFO[0960] TCP connection established                    local_addr="10.120.147.140:56018" remote_addr="pulsar://pulsar-broker.pulsar1.svc.cluster.local:6650"
INFO[0960] Connection is ready                           local_addr="10.120.147.140:56018" remote_addr="pulsar://pulsar-broker.pulsar1.svc.cluster.local:6650"


And the server has a shedding performed.

Since it is very costy to have the DEBUG level log turned on, I didn't have the chance to catch debug level messages.

Reproducing the issue

I've written two parts that can reproduce such issue.
Producer that would delivery messages with variadic delay (from 10s to 1h).
Consumer that would receive messages.

Wait for the message cumulate until the expected number, the consumer hangs with very little message received.

Additional information

It might relates to the setting of managedLedgerMaxUnackedRangesToPersist but for our usage type, it is not possible to increase this setting infinitely because the message would grow.
Also I've notice that when the individuallyDeletedMessages is quite big, every time a consumer reconnect to the broker would cause both broker and zookeeper to have a peak CPU usage, I assume it is because pulsar was trying to compute the actual messages that shall be dispatched.
I wonder if there's a way to optimize such issue or a way to tune it ? Or this is not the correct way of using pulsar ?

Are you willing to submit a PR?

• I'm willing to submit a PR!

[lhotari]

There's some tuning advice in #23990 .

Are you already using delayedDeliveryTrackerFactoryClassName=org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory to get the benefits of PIP-195: New bucket based delayed message tracker?

[YanshuoH]

I do try to reference the discussion #23990 to tune the delayed messages.

Forgot to paste my configuration:

  managedLedgerMaxUnackedRangesToPersist: "200000"
  managedLedgerMaxBatchDeletedIndexToPersist: "200000"
  managedLedgerPersistIndividualAckAsLongArray: "true"
  managedLedgerUnackedRangesOpenCacheSetEnabled: "true"
  persistentUnackedRangesWithMultipleEntriesEnabled: "true"
  delayedDeliveryTrackerFactoryClassName: "org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory"

For example, with these internal stats:

      "numberOfEntriesSinceFirstNotAckedMessage": 718613,
      "totalNonContiguousDeletedMessagesRange": 143483,

My testing client can always confirm the slow dispatch:

2025/12/01 18:26:27 received 482506 messages, 209 failures
2025/12/01 18:26:42 received 482507 messages, 210 failures
2025/12/01 18:26:57 received 482509 messages, 216 failures

As you can see, the number of entries is suffit enough to deliver the message as my messages are with delay from 10s to 1h.
And one strange thing is that the numberOfEntriesSinceFirstNotAckedMessage stays still even with more messages produced.
Yet the message consumption is quite very slow with a few messages dispatched and most of the failures are receive timeout.

Also I'd like to provide my client config:

	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:                       c.Topic,
		SubscriptionName:            c.Subscription,
		Type:                        pulsar.Shared,
		SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
		ReceiverQueueSize:           1,
		Name:                        uuid.NewString(),
		AckWithResponse:             true,
	})

[lhotari]

I do try to reference the discussion #23990 to tune the delayed messages.

Forgot to paste my configuration:

  managedLedgerMaxUnackedRangesToPersist: "200000"
  managedLedgerMaxBatchDeletedIndexToPersist: "200000"
  managedLedgerPersistIndividualAckAsLongArray: "true"
  managedLedgerUnackedRangesOpenCacheSetEnabled: "true"
  persistentUnackedRangesWithMultipleEntriesEnabled: "true"
  delayedDeliveryTrackerFactoryClassName: "org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory"

You seem to be missing the compression configuration

managedCursorInfoCompressionType=LZ4
managedLedgerInfoCompressionType=LZ4

you might also want to increase managedLedgerMaxUnackedRangesToPersist to a value that won't overflow.

Also I'd like to provide my client config:

	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:                       c.Topic,
		SubscriptionName:            c.Subscription,
		Type:                        pulsar.Shared,
		SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
		ReceiverQueueSize:           1,
		Name:                        uuid.NewString(),
		AckWithResponse:             true,
	})

Have you tested with a larger receiver queue size? The default value is 1000.

[YanshuoH]

1. For the compression level, yes I've added it but it seems not having much effect.
2. As for the receiver queue size, my experience is that for delay message that if we want the message to arrive on time, it is necessary to have the receiver queue size to be 1, yet even though we change the receiver queue size to 1000, it does not change the issue.
3. Back to the managedLedgerMaxUnackedRangesToPersist, my former setting is 100,000, now it is 200,000. And I can confirm that setting this value to a bigger threshold works for a while. But my question is that the business keeps growing and the holes keeps growing to. I wonder what will happen if we keep increasing the managedLedgerMaxUnackedRangesToPersist ? That is my original question about:

I wonder if there's a way to optimize such issue or a way to tune it ? Or this is not the correct way of using pulsar ?

[lhotari]

1. For the compression level, yes I've added it but it seems not having much effect.
2. As for the receiver queue size, my experience is that for delay message that if we want the message to arrive on time, it is necessary to have the receiver queue size to be 1, yet even though we change the receiver queue size to 1000, it does not change the issue.

Thanks for checking. The purpose was to find out whether the behavior is any different. For production use, it's also recommended to enable compression.

4. Back to the managedLedgerMaxUnackedRangesToPersist, my former setting is 100,000, now it is 200,000. And I can confirm that setting this value to a bigger threshold works for a while. But my question is that the business keeps growing and the holes keeps growing to. I wonder what will happen if we keep increasing the managedLedgerMaxUnackedRangesToPersist ? That is my original question about:>

The max number of "ack holes" (== "unacked ranges") is "number_of_entries_since_oldest_non_acked_message / 2".
If this exceeds the limit managedLedgerMaxUnackedRangesToPersist, the messages could get redelivered after a broker restart or when the namespace bundle gets moved to another broker because of load shedding / load balancing.

I wonder if there's a way to optimize such issue or a way to tune it ? Or this is not the correct way of using pulsar ?

For the acknowledgment state, the way to optimize is to categorize messages with similar delays to the same topic. On the consumer side, you can consume from multiple topics. This reduces the state size on the broker.

For optimizing BucketDelayedDeliveryTrackerFactory performance, @Denovo1998 has been working on #24739.

[Denovo1998]

@YanshuoH
In PR #24739:

• replaces StampedLock/synchronized with ReentrantReadWriteLock and narrows the write‑lock sections in addMessage / getScheduledMessages,
• moves snapshot persistence/loading to asynchronous code paths so that tracker operations do not block on I/O, and
• decouples the tracker from the dispatcher to avoid lock‑ordering issues and make thread‑safety better testable.

These changes are intended to reduce lock contention and avoid the tracker itself becoming a bottleneck when there are lots of delayed messages.

---

However, your issue also clearly shows very large numberOfEntriesSinceFirstNotAckedMessage and totalNonContiguousDeletedMessagesRange. That points to a big part of the cost coming
from the ManagedLedger acknowledgment state (many “ack holes”) and the configured limit managedLedgerMaxUnackedRangesToPersist. PR #24739 does not change how that ack state is
represented or persisted, so it is not expected to fully solve the behavior you’re seeing by itself.

In addition to the tuning that has already been discussed (compression, receiverQueueSize, etc.), the most promising mitigation on the usage side seems to be what @lhotari already
suggested:

• grouping messages with similar delays into separate topics, so that each topic maintains a smaller ack state; the consumer can subscribe to multiple topics to preserve the overall behavior.

For the current implementation mechanism of delayed messages in pulsar, this may be the only solution.

[Denovo1998]

By the way. I have a speech this Saturday in China about Pulsar delayed messages, so I can briefly mention it here.

There are mainly two implementation methods for delayed messages:

---

1. Delayed Production

This means that after the Broker receives a message, it first stores the message in some storage (such as a built-in Topic), and there is an independent thread to determine whether the data has expired. If the data expires, the message is then written to the actual Topic, allowing the consumer to consume the data normally.

Advantages: Minimal changes to the main production and consumption flow. Only need to differentiate delayed messages when writing, write them to temporary storage, and independently implement timed detection to write expired data to the target Topic.

Disadvantages: When the message volume is large, the message expiration time may not be accurate.

From industry implementations, most choose this method.

2. Delayed Visibility

Each time consumption occurs, check if there are expired delayed messages. If yes, pull the delayed messages from temporary storage and return them to the consumer. Delayed messages that have not expired are filtered out and kept invisible.

In this sub-scheme, when the producer writes data, it generally also writes the data to temporary storage. But unlike the previous method, each time consumption actively checks whether there are messages expired in the third-party storage. If so, the expired data is returned to the client.

Advantages: Eliminates the need for timed thread detection and writing logic, simplifying the flow significantly. Moreover, this method makes delayed messages highly accurate.

Disadvantages: The QPS of consumption operations is usually very high, so this temporary storage must have very high performance for retrieval operations, and memory usage cannot be too high. Additionally, checking for delayed messages every time may cause performance issues.

---

In Pulsar, it is implemented through method two, delayed visibility (with some differences), and as with Pulsar's ack design, it affects normal consumption and production.

So I propose a hybrid architecture.

There is such an external service (although it is an open-source project under Apache-2.0 license, I won't mention its name here), which is implemented using the delayed production method. The client sends an "add/cancel" delayed message command (including the message payload). After receiving this command from an internal topic, it persists the message to be triggered in the future to RocksDB, scans for expiration by the second, and then delivers it to the target Topic.

We can hand over delayed messages that require "complex delay capabilities" (can be canceled, can have recurring delays) and have long delay times (day-level or even month-level) to this external service for processing. If high-precision second-level or minute-level delayed messages are needed, continue using Pulsar's internal implementation (recommended to use InMemory implementation). I think such a hybrid architecture can well solve many performance issues with delayed messages in Pulsar currently.

[oneby-wang]

Hi, @YanshuoH, based on @lhotari and @Denovo1998 suggestions, I would like to recommend you to use MessageRoutingMode.CustomPartition and define your custom routing logic in producer. In this way, you don't have to change your producer and consumer logic.

First, your delayed topic must be a partitioned-topic, assume it has 60 partitions(let's call it n), so you can seperate your delay time range to 1 minute(let's call it t), each partition will have at most 1min ack holes(the large ack holes problem is solved).

Second, define your custom routing logic. You can route 0 ~ 1min messages to partition-0, route 1 ~ 2min messages to partition-1, ......, route 59 ~ 60min messages to partition-59.

You can tune n and t to custom your routing logic without modifying producer and consumer logic.

[Denovo1998]

+1

[sammyne]

Delayed Visibility

Are there any PR implementing this two methods? Or they are just not implemented yet? @Denovo1998

[Denovo1998]

@sammyne
This Saturday, I will mention how to transform and integrate the external delayed message service based on RocksDB in a meetup in China. ( At present, I have personally started to do integrated pr. It is expected that a complete delayed message hybrid architecture service can be made in the next meetup next year)

But this kind of external service has nothing to do with Pulsar itself. I understand that this kind of pr will not be submitted and merged in Pulsar‘s official library.

[YanshuoH]

Thanks guys, appreciate your responses.

Now I fully understand our current status and may implement your suggestions.

Just for your information, I previously used AWS SQS (since our services are on AWS) and I implemented the delayed message as Denovo1998 mentioned:

2. Delayed Visibility

The main constraint is that AWS SQS only support max 15 minutes delay in FIFO queues and as I've addressed, our service may require a variadic delay from 1m to 1h. So the consumer would receive and resend a new message with proper delay. But that's a digression.

Close this issue for now and I'll keep looking at pulsar's new updates.

Thank you again.