[fix][broker] Fix delayed messages stalling with isDelayedDeliveryDeliverAtTimeStrict=true#26012
Merged
lhotari merged 4 commits intoJun 12, 2026
Conversation
…iverAtTimeStrict=true ### Motivation With `isDelayedDeliveryDeliverAtTimeStrict=true`, delayed messages can remain undelivered indefinitely past their `deliverAt` time while a consumer is blocked in `receive()`. The stalled messages are only released when an unrelated dispatch event happens (e.g. a new publish or a consumer reconnect); a quiet topic stalls forever. With the default `strict=false` the same traffic is delivered on time. Root cause is in `AbstractDelayedDeliveryTracker.updateTimer()`. Because timestamps are trimmed for memory efficiency (up to ~511ms with the default `tickTimeMillis=1000`), `getScheduledMessages()` can pop a message slightly before its real `deliverAt`. In strict mode the dispatcher re-adds the not-yet-due message, which calls `updateTimer()`: 1. The existing timer (armed for the next message) is cancelled. 2. `delayMillis` for the re-added message is negative, so the method takes its early return -- but it leaves `currentTimeoutTarget` pointing at the previous target and `timeout` non-null (now cancelled). 3. When the early message is finally delivered, the next `updateTimer()` sees `timestamp == currentTimeoutTarget`, concludes the timer is already correctly armed, and returns. No live timer exists, so the remaining messages are never delivered until an external dispatch round finds them. Fixes apache#25996 ### Modifications - In the `delayMillis < 0` early return of `updateTimer()`, reset `currentTimeoutTarget = -1` and `timeout = null` so a later call cannot short-circuit on stale state and will re-arm the timer. - Add a deterministic unit test `testStrictModeTimerStallsAfterEarlyPopAndReAdd` that reproduces the early-pop/re-add sequence and asserts a delivery timer remains armed for the still-pending message. It fails without the fix and passes with it. ### Verifying this change This change is covered by the added unit test. Existing `InMemoryDeliveryTrackerTest` and `BucketDelayedDeliveryTrackerTest` still pass. Assisted-by: Claude Code (Opus 4.8)
lhotari
requested changes
Jun 12, 2026
lhotari
left a comment
lhotari
left a comment
Member
There was a problem hiding this comment.
Root cause is in AbstractDelayedDeliveryTracker.updateTimer(). Because delivery timestamps are trimmed for memory efficiency (up to ~511ms with the default tickTimeMillis=1000), getScheduledMessages() can pop a message slightly before its real deliverAt.
The changes itself are fine, but I think that this root cause should be addressed. It causes unnecessary re-adding which causes a performance overhead.
Member
I'll push changes to this PR to address the root cause about unnecessary re-adding. |
…mode delivery accuracy Follow-up to the apache#25996 fix in the previous commit. Thread safety of the delivery timer state in AbstractDelayedDeliveryTracker: - Guard timeout, currentTimeoutTarget and lastTickRun with a dedicated timeoutLock shared by updateTimer(), rescheduleTimer(), run() and close(). Previously the timer thread and dispatcher threads mutated these fields without a common lock, and BucketDelayedDeliveryTracker manipulated the timeout field directly under a different lock (its own monitor). - Snapshot getNumberOfDelayedMessages()/nextDeliveryTime() before acquiring timeoutLock in updateTimer(), keeping timeoutLock a leaf lock that cannot participate in lock-order inversion with subclasses that synchronize those methods on the tracker instance. - In run(), only clear the timer state when the triggered timeout is the currently armed one. A superseded timeout can still fire after losing the race with its cancellation; clearing the newer timer's state made the next updateTimer() call arm a duplicate timer. - Add rescheduleTimer() so BucketDelayedDeliveryTracker no longer mutates the timeout field directly. - Generalize the apache#25996 stale-state fix: reset currentTimeoutTarget on every cancellation path in updateTimer() and also in close(). - Make tickTimeMillis volatile: it is updated via resetTickTime() from dispatcher threads and read on the timer thread. Strict-mode delivery accuracy in InMemoryDelayedDeliveryTracker: - With isDelayedDeliveryDeliverAtTimeStrict=true, round the bucketed delivery timestamp up to the bucket boundary instead of down. The memory-saving timestamp trimming made messages visible to the dispatcher up to ~511ms (with the default tickTimeMillis=1000) before their deliverAt time; the dispatcher would put the not-yet-due message back into the tracker and re-trigger reads in a loop until the deliverAt time was reached. With round-up, a bucket becomes due only once every deliverAt time inside it has passed: messages are never delivered early and at most one bucket (less than tickTimeMillis) late, within the documented delay of tickTimeMillis plus the timer granularity. Non-strict mode behavior is unchanged. Tests: - Replace the apache#25996 regression test with testStrictModeNeverDeliversEarlyAndKeepsTimerArmed, covering the new strict-mode semantics and the original timer stall. - Add testStaleTimerTriggerDoesNotClearNewerTimer covering the stale timer trigger race in run().
…DelayedDeliveryTracker - Use addLong(long) instead of the varargs add(long...), which allocated a single-element long[] for every message added to the tracker. - Keep the bitmap cardinality as a long and only cast to int in the branch where the comparison against the remaining maxMessages budget guarantees it fits, instead of truncating the long cardinality up front. - When a per-ledger bitmap holds more entries than the remaining maxMessages budget in getScheduledMessages(), stream only the n lowest entry ids and remove them with a single bulk andNot(), instead of materializing the whole bitmap with toArray() and removing the selected entries one by one. - Add testGetScheduledMessagesWithMaxMessagesSmallerThanBucket to cover the partial bitmap drain (cardinality > n), which had no test coverage: the lowest entry ids are returned first, repeated calls continue without duplicates, and the tracked message count stays consistent.
Member
|
I also added a performance improvement to |
11 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #25996
Motivation
With
isDelayedDeliveryDeliverAtTimeStrict=true, delayed messages can remain undelivered indefinitely past theirdeliverAttime while a consumer is blocked inreceive(). The stalled messages are only released when an unrelated dispatch event happens (e.g. a new publish or a consumer reconnect); on a quiet topic the delay is unbounded. With the defaultstrict=falsethe same traffic is delivered on time.Root cause is in
AbstractDelayedDeliveryTracker.updateTimer(). Because delivery timestamps are trimmed for memory efficiency (up to ~511ms with the defaulttickTimeMillis=1000),getScheduledMessages()can pop a message slightly before its realdeliverAt. In strict mode the dispatcher re-adds the not-yet-due message, which callsupdateTimer():delayMillisfor the re-added message is negative, so the method takes its early return — but it leavescurrentTimeoutTargetpointing at the previous target andtimeoutnon-null (now cancelled).updateTimer()seestimestamp == currentTimeoutTarget, concludes the timer is already correctly armed, and returns. No live timer exists, so the remaining delayed messages are never delivered until an external dispatch round happens to find them.strict=falseis immune because its cutoff (now + tickTimeMillis) covers the trim window, so early-popped messages are delivered instead of being re-added.Modifications
delayMillis < 0early return ofAbstractDelayedDeliveryTracker.updateTimer(), resetcurrentTimeoutTarget = -1andtimeout = nullso a later call cannot short-circuit on stale state and will correctly re-arm the timer.testStrictModeTimerStallsAfterEarlyPopAndReAddinInMemoryDeliveryTrackerTestthat reproduces the early-pop / re-add sequence and asserts a delivery timer remains armed for the still-pending message. It fails without the fix and passes with it.Verifying this change
This change is already covered by the added unit test
testStrictModeTimerStallsAfterEarlyPopAndReAdd. ExistingInMemoryDeliveryTrackerTestandBucketDelayedDeliveryTrackerTestcontinue to pass.Documentation
doc-requireddoc-not-neededMatching PR in forked repository
PR in forked repository: coderzc/pulsar (this change is a bug fix; the fix branch is
fix/delayed-delivery-strict-timer-stall-25996).