[fix][broker] Fix bucket delayed delivery state consistency issues

[nodece]

Motivation

Fix permanent message loss during bucket snapshot merge, and fix counter / delivery correctness issues that emerge once the data-loss fix is applied.

• Merge permanently loses segment 1 data. getRemainSnapshotSegment loads from currentSegmentEntryId + 1 (segment 2+), skipping segment 1. After merge, old snapshots are deleted from BookKeeper. On restart, recovery rebuilds an incomplete bitMap — messages in segment 1 can never be delivered.

• Loading segment 1 during merge creates duplicate queue entries. Segment 1 messages are already in sharedBucketPriorityQueue from the initial seal; merge pushes them in again. The existing pop path ignores removeIndexBit's return value, so both pops deliver and decrement the counter.

• Expired re-add leaks counter. addMessage with deliverAt <= cutoffTime returns false without removing the existing bit. The bit stays and the counter remains inflated.

Modifications

• Fix merge to load all segments from segment 1

  getAllSnapshotSegments now loads segment 1 … lastSegmentEntryId instead of currentSegmentEntryId+1 … lastSegmentEntryId. The merged BookKeeper snapshot preserves complete bucket state.

• Extract BucketDelayedMessageIndex as the single runtime truth source

  The runtime dedup state is consolidated into BucketDelayedMessageIndex which owns the bitmap and counter together. The counter is an invariant of the bitmap cardinality, not something callers must manually keep in sync.

  ImmutableBucket.delayedIndexBitMap is now a frozen snapshot used only for BookKeeper writes, merge composition (OR of source buckets), and recovery. It is no longer consulted at runtime for dedup or counter.

  Pop path now checks the return value:

  sharedBucketPriorityQueue.pop();
  if (removeIndexBit(ledgerId, entryId)) {   // false → duplicate, skip
      positions.add(PositionFactory.create(ledgerId, entryId));
      --n;
  }
  

  First pop of a position delivers and decrements; subsequent pops are silently skipped.

• Fix expired re-add cleanup

  addMessage with deliverAt <= cutoffTime now calls removeIndexBit before returning false.

• Recovery restores into runtime index

  index.restore(bucket.getDelayedIndexBitMap()) populates the runtime index so all recovered messages participate in delivery dedup.

• Trim is index-aware

  deleteBucketSnapshot iterates the trimmed bucket's frozen bitMap and calls index.untrack per bit. Only bits still in the runtime index decrement the counter.

• Remove Bucket base class

  ImmutableBucket's inherited delayedIndexBitMap served a dual role (runtime dedup + BK snapshot) that made these bugs hard to reason about. ImmutableBucket and MutableBucket are now independent classes sharing a BucketContext record.

Semantics

Pulsar's existing at-least-once delivery semantics are preserved. After merge + restart, messages that were dispatched but whose acknowledgement was not persisted before restart may be replayed. Actual delivery goes through the cursor's markDeletedPosition check — already-acked messages are filtered by the dispatcher. The only window for re-delivery is the same one that exists for non-delayed messages.

[nodece]

BucketDelayedDeliveryTest.testDelayedDeliveryWithMultipleConcurrentReadEntries is a flaky test: #19902

[void-ptr974]

When a selected bucket still has its current segment loaded in sharedBucketPriorityQueue, reading from segment 1 reads that same segment from storage again. The merge then creates a new bucket; creating a bucket puts the first segment of its new snapshot into sharedBucketPriorityQueue, but the entries already queued for the source buckets are not removed.

For example, assume a source bucket's current segment contains 1:1 with a delivery time of 10ms. Before the merge, one copy of 1:1 is already in the shared queue. The merged snapshot also contains 1:1, and its first segment enqueues another copy. At 10ms, getScheduledMessages(1) can return 1:1; the next call can return 1:1 again. The second dequeue also decrements numberDelayedMessages again, so the count becomes incorrect.

If currentSegmentEntryId has advanced, reading all segments also includes older segments that have already been dispatched. Those entries can be persisted in the merged snapshot and replayed after a restart.

We need to keep the current segment recoverable without adding a second live queue entry, and exclude entries that have already been dispatched.

[nodece]

@void-ptr974 This PR body has been updated.

[void-ptr974]

Thanks for the update. The main correctness issues are addressed. LGTM.