Skip to content

[fix][broker] Fix snapshot creation race causing delayed delivery bucket trim failures - #26260

Open
void-ptr974 wants to merge 1 commit into
apache:masterfrom
void-ptr974:fix/bucket-trim-create-race
Open

[fix][broker] Fix snapshot creation race causing delayed delivery bucket trim failures#26260
void-ptr974 wants to merge 1 commit into
apache:masterfrom
void-ptr974:fix/bucket-trim-create-race

Conversation

@void-ptr974

@void-ptr974 void-ptr974 commented Aug 1, 2026

Copy link
Copy Markdown
Contributor

Motivation

BucketDelayedDeliveryTracker publishes a newly sealed immutable bucket before its asynchronous snapshot creation completes. If the same addMessage call causes the bucket count to exceed delayedDeliveryMaxNumBuckets, trim can immediately select a bucket whose snapshot ID is not available yet. This makes the current trim pass fail and delays cleanup until a later trim is triggered.

Modifications

Consider a tracker with a maximum of 5 buckets and a first active ledger ID of 50:

  1. Rapid message additions create 6 immutable buckets while their snapshot creations are still in flight. The same addMessage call that creates the sixth bucket starts trim because the limit has been exceeded.
  2. Since the bucket ranges end before ledger 50, trim considers them orphaned and attempts to delete them sequentially.
  3. Before this change, deletion starts immediately. For an in-flight snapshot, the bucket ID is neither cached in the bucket nor recorded in the cursor properties, so getAndUpdateBucketId() calls Long.parseLong(null) and throws NumberFormatException. The trim chain stops at that failure and trimFuture completes exceptionally. It is not permanently pending, but the failed deletion is left for a later trim attempt.
  4. With this change, each deletion is chained from the bucket's snapshot creation future. After creation completes, trim revalidates under the tracker lock that the bucket ID is valid, the range is still before the current first active ledger, and the exact range still maps to the same bucket instance.
  5. Only a bucket that still passes those checks is deleted. After deletion succeeds, the mapping is checked again before removing the bucket indexes and updating the delayed-message count, preventing an older asynchronous continuation from modifying newer tracker state.

The wait is asynchronous and does not block the dispatcher thread.

Verifying this change

Added deterministic tests that hold snapshot creation in flight and verify that:

  • Trim does not delete snapshots before creation completes and resumes after creation is released.
  • Trim revalidates bucket eligibility after waiting and does not delete based on stale state.

Verified with:

  • ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTrackerTest
  • ./gradlew :pulsar-broker:test --tests org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTrackerThreadSafetyTest
  • ./gradlew quickCheck

Does this pull request potentially affect one of the following parts:

  • Dependencies (add or upgrade a dependency)
  • The public API
  • The schema
  • The default values of configurations
  • The threading model
  • The binary protocol
  • The REST endpoints
  • The admin CLI options
  • The metrics
  • Anything that affects deployment

@void-ptr974 void-ptr974 changed the title [fix][broker] Wait for delayed delivery snapshot creation before trimming [fix][broker] Fix snapshot creation race causing delayed delivery bucket trim failures Aug 1, 2026
@void-ptr974
void-ptr974 marked this pull request as ready for review August 1, 2026 14:25
}
}
return null;
return bucket.asyncDeleteBucketSnapshot(stats).thenRun(() -> {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The orphan eligibility is validated only before the asynchronous snapshot deletion begins. Once this synchronized block is released, the cursor may be reset or its mark‑deleted position could move backward while asyncDeleteBucketSnapshot() is still running.

In such cases, the completion block only rechecks the range‑to‑bucket identity, so it may still delete bucket state even if range.upperEndpoint() < firstActiveLedgerId() no longer holds. The existing test updates firstLedgerId before the snapshot‑create future is released, but it does not cover modifications after this validation and after storage deletion has started.

Could we add a deletion ownership or version guard that spans the duration of the asynchronous operation, or base orphan cleanup on a monotonic physical‑ledger boundary? Please also include a test that blocks deleteBucketSnapshot(), changes firstLedgerId after deletion begins, and then completes the delete.

.log("Failed to delete bucket snapshot");
throw new CompletionException(t);
}
return bucket.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This ensures the global trimFuture waits for every selected bucket's snapshot-creation future. If any create future remains pending for an extended period, trimFuture.isDone() stays false, which prevents later addMessage() calls from triggering another trim/merge and also chains clear() behind the same future.

Would it be safer to skip buckets still in the CREATING state and retrigger trim when their creation finishes, rather than holding the global trim/merge gate? At minimum, it would be helpful to document the bounded-completion guarantee or add a test for a create future that never completes.

@nodece

nodece commented Aug 6, 2026

Copy link
Copy Markdown
Member

@void-ptr974 #26280 should include this fix. When fixing the trim and segement loding, I found this issue.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants