[fix][broker] Check deliverAt before containsMessage in bucket addMessage - #26230

@nodece Thanks for the fix — the reordering itself looks right (it matches the InMemoryDelayedDeliveryTracker semantics, where expiry is decided before dedup). One follow-up observation from a local review (Claude Fable 5 + Codex gpt-5.6-sol — both flagged this independently):

When addMessage now returns false for an expired message that is still tracked, the entry is not removed from the tracker. If the position is still in sharedBucketPriorityQueue (e.g. the dispatcher re-reads an already-due tracked message during replay/rewind), the dispatcher delivers it immediately, and getScheduledMessages() later returns the same position again when it pops it from the queue — a duplicate-delivery window. numberDelayedMessages also transiently overcounts until the stale entry pops.

Making the false path remove the entry isn't a one-liner, though: TripleLongPriorityQueue has no positional removal (only head pop), and clearing just the bitmap bit wouldn't be enough because the pop path adds the position and decrements the counter unconditionally:

A complete fix would clear the index bit in addMessage before returning false and make the pop path skip positions whose bit is already gone, mirroring the existing firstLiveLedgerId branch:

Given Pulsar's at-least-once semantics this may well be acceptable as-is (an occasional redelivery beats the pre-fix indefinite skip), but it seemed worth recording — and possibly worth handling together with the dispatcher-side expiration check mentioned as follow-up work in the PR description.

@lhotari Good catch. I only considered the case where the expired message could be skipped indefinitely, but I missed the potential duplicate-delivery window caused by leaving the stale entry in the priority queue.

The current Bucket implementation has become quite complicated over time — different bucket types, index maintenance, storage lifecycle, and message counters are mixed together, which makes these edge cases harder to reason about.

I'm currently refactoring this part and will try to address this issue as part of the redesign. The direction is to make the index state and queue state more consistent, and avoid relying on implicit cleanup during pop operations.