Skip to content

[fix][broker] Clear delayed delivery state before resetting the cursor #26420

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nodece wants to merge 1 commit into
base: master
Choose a base branch
from
+116 −76
Open

[fix][broker] Clear delayed delivery state before resetting the cursor #26420

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 71 additions & 76 deletions ...ker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -983,88 +983,83 @@ private CompletableFuture<Void> resetCursorInternal(Position finalPosition, Comp
}
}

disconnectFuture.whenComplete((aVoid, throwable) -> {
if (dispatcher != null) {
dispatcher.resetCloseFuture();
}

if (throwable != null) {
log.error()
.exception(throwable)
.log("Failed to disconnect consumer from subscription");
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
future.completeExceptionally(
new SubscriptionBusyException("Failed to disconnect consumers from subscription"));
return;
}

log.info()
.log("Successfully disconnected consumers from subscription, proceeding with cursor reset");

CompletableFuture<Boolean> forceReset = new CompletableFuture<>();
if (topic.getTopicCompactionService() == null) {
forceReset.complete(false);
} else {
topic.getTopicCompactionService().getLastCompactedPosition().thenAccept(lastCompactedPosition -> {
Position resetTo = finalPosition;
if (lastCompactedPosition != null && resetTo.compareTo(lastCompactedPosition.getLedgerId(),
lastCompactedPosition.getEntryId()) <= 0) {
forceReset.complete(true);
} else {
disconnectFuture
.thenCompose(__ -> {

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

[QUALITY] the whenComplete -> thenCompose rewrite silently dropped the disconnect-failure branch (resetCloseFuture skipped, 412 becomes 500) — currently unreachable, flagging so the deletion is deliberate

Moving from whenComplete to thenCompose changed what happens when disconnectFuture completes exceptionally. The base revision ran, on either outcome:

if (dispatcher != null) { dispatcher.resetCloseFuture(); }   // unconditional
if (throwable != null) {
    ...
    future.completeExceptionally(
        new SubscriptionBusyException("Failed to disconnect consumers from subscription"));
    return;
}

thenCompose skips its body when the upstream future fails, so resetCloseFuture() no longer runs — leaving a stale closeFuture on the dispatcher (PersistentDispatcherMultipleConsumers.java:657-659) — and the terminal .exceptionally surfaces BrokerServiceException instead. That would change the admin response: SubscriptionBusyException maps to 412 PRECONDITION_FAILED (PersistentTopicsBase.java:2816-2822 and :2524-2530), while anything else falls through resumeAsyncResponseExceptionally to 500.

This is currently unreachable, so it is a latent semantics change rather than a live bug: in every persistent dispatcher closeFuture is only ever completed with complete(null) (PersistentDispatcherMultipleConsumers.java:631-641, PersistentDispatcherMultipleConsumersClassic.java:541-553, AbstractDispatcherSingleActiveConsumer.java:337-343). I am flagging it only so the removal is deliberate rather than incidental to the reflow — if the branch is genuinely dead, dropping it is fine, but it would be good to say so in the PR description.

log.info()
.log("Successfully disconnected consumers from subscription, proceeding with cursor reset");
if (dispatcher != null) {
dispatcher.resetCloseFuture();
return dispatcher.clearDelayedMessages();

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

[INTENT MISMATCH] clear() awaits in-flight trims but not an in-flight bucket segment load, so the "in-flight loads could race" motivation is only partly addressed

The second motivation bullet says the fix stops "in-flight trims/loads/deletes from before the reset" racing the replayed state. Awaiting clearDelayedMessages() here covers trims and deletes, but not loads.

BucketDelayedDeliveryTracker.clear() (:795-816) chains only off trimFuture. The in-flight segment load tracked in pendingLoad (declared at :131, assigned at :717) is never awaited, and its continuation (:718-734) re-populates snapshotSegmentLastIndexMap and sharedBucketPriorityQueue. Both that continuation and the body of clear() (:806) take the tracker monitor, so they are mutually exclusive but unordered.

Failure scenario. A bucket segment load is already in flight when the reset disconnects consumers. clear() empties the queues and cleanImmutableBuckets() removes the immutable buckets; the outstanding load's continuation then re-inserts pre-reset indexes into sharedBucketPriorityQueue and re-registers a bucket that was just removed. asyncResetCursor proceeds against a tracker that is not actually empty, which is the state this PR set out to prevent.

To be fair on scope: this is a pre-existing property of clear() (shared with clearBacklog and unsubscribe), not a regression introduced here, and the window is narrow — consumers are already disconnected by this point, so only a load started before the reset can land. But since the stated purpose of the change is to make the reset wait for the delayed state to settle, it is worth either awaiting pendingLoad in clear() or narrowing the second motivation bullet.

}
return CompletableFuture.completedFuture(null);

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

[BUG] the dispatcher == null branch leaves pre-reset bucket snapshots on the cursor, so the PR's third motivation bullet is not fixed for that case

This branch returns without cleaning anything, so a reset on a subscription with no dispatcher leaves the pre-reset #pulsar.internal.delayed.bucket* properties on the cursor — the third motivation bullet ("the bucket cursor properties stayed on the cursor, so a later tracker recovery would load pre-reset buckets") is not fixed for this case.

The asymmetry is inside this very method. When a dispatcher does exist but has no tracker yet, clearDelayedMessages() falls back to cleanResidualSnapshots(cursor) (PersistentDispatcherMultipleConsumers.java:1387-1402), which deletes the snapshots and removes the cursor properties (BucketDelayedDeliveryTrackerFactory.java:115-132). With no dispatcher at all — the same "no tracker" state — nothing runs. The unsubscribe path the description cites as precedent handles this explicitly (PersistentTopic.java:1394-1410).

Reachability. PersistentSubscription.dispatcher is never assigned null after construction (it is only created/reused in addConsumer), so dispatcher == null means "no consumer has connected since the topic was loaded" — the state right after a broker restart or a topic unload. internalResetCursorOnPosition has no guard requiring a connected consumer (PersistentTopicsBase.java:2750-2800), and resetting a cursor with consumers stopped is a common operational sequence.

Failure scenario. Bucket delayed delivery enabled; a Shared subscription has tracked delayed messages and persisted bucket snapshots; the topic is unloaded; with no consumer connected an operator resets the cursor forward. The snapshots survive the reset, and the first consumer to connect rebuilds the tracker from them (BucketDelayedDeliveryTracker.java:181-199) without filtering against the new mark-delete position — recovering already-skipped entries, inflating the delayed-message count, and retaining bucket snapshot ledgers that nothing will delete.

The description asserts this branch is safe ("the recovered bucket snapshots stay valid for the replay ... re-added messages dedup through the index bitmap"). That argument holds for a backward reset, but not for a forward one. Either route dispatcher == null through cleanResidualSnapshots(cursor) the way unsubscribe does, or narrow the claim in the description to backward resets and say why forward resets are acceptable.

})
.thenCompose(__ -> {
CompletableFuture<Boolean> forceReset = new CompletableFuture<>();
if (topic.getTopicCompactionService() == null) {
forceReset.complete(false);
} else {
topic.getTopicCompactionService().getLastCompactedPosition()
.thenAccept(lastCompactedPosition -> {
Position resetTo = finalPosition;
if (lastCompactedPosition != null
&& resetTo.compareTo(lastCompactedPosition.getLedgerId(),
lastCompactedPosition.getEntryId()) <= 0) {
forceReset.complete(true);
} else {
forceReset.complete(false);
}
}).exceptionally(ex -> {
forceReset.completeExceptionally(ex);
return null;
});
}
}).exceptionally(ex -> {
forceReset.completeExceptionally(ex);
return null;
});
}

forceReset.thenAccept(forceResetValue -> {
cursor.asyncResetCursor(finalPosition, forceResetValue, new AsyncCallbacks.ResetCursorCallback() {
@Override
public void resetComplete(Object ctx) {
log.debug()
.attr("finalPosition", finalPosition)
.log("Successfully reset subscription to position");
if (dispatcher != null) {
dispatcher.cursorIsReset();
dispatcher.afterAckMessages(null, finalPosition);
return forceReset;
})
.thenAccept(forceResetValue -> {
cursor.asyncResetCursor(finalPosition, forceResetValue, new AsyncCallbacks.ResetCursorCallback() {
@Override
public void resetComplete(Object ctx) {
log.debug()
.attr("finalPosition", finalPosition)
.log("Successfully reset subscription to position");
if (dispatcher != null) {
dispatcher.cursorIsReset();
dispatcher.afterAckMessages(null, finalPosition);
}
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
future.complete(null);
}
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
future.complete(null);
}

@Override
public void resetFailed(ManagedLedgerException exception, Object ctx) {
log.error()
.attr("finalPosition", finalPosition)
.exception(exception)
.log("Failed to reset subscription to position");
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
// todo - retry on InvalidCursorPositionException
// or should we just ask user to retry one more time?
if (exception instanceof InvalidCursorPositionException) {
future.completeExceptionally(new SubscriptionInvalidCursorPosition(exception.getMessage()));
} else if (exception instanceof ConcurrentFindCursorPositionException) {
future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
} else {
future.completeExceptionally(new BrokerServiceException(exception));
@Override
public void resetFailed(ManagedLedgerException exception, Object ctx) {
log.error()
.attr("finalPosition", finalPosition)
.exception(exception)
.log("Failed to reset subscription to position");
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
// todo - retry on InvalidCursorPositionException
// or should we just ask user to retry one more time?
if (exception instanceof InvalidCursorPositionException) {
future.completeExceptionally(
new SubscriptionInvalidCursorPosition(exception.getMessage()));
} else if (exception instanceof ConcurrentFindCursorPositionException) {
future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
} else {
future.completeExceptionally(new BrokerServiceException(exception));
}
}
}
});
}).exceptionally((e) -> {
log.error()
.exception(e)
.log("Error while resetting cursor");
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
future.completeExceptionally(new BrokerServiceException(e));
return null;
});
}).exceptionally((e) -> {
log.error()
.exception(e)
.log("Error while resetting cursor");
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
future.completeExceptionally(new BrokerServiceException(e));
return null;
});
});
return future;
}

Expand Down
45 changes: 45 additions & 0 deletions .../src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,51 @@ public void testBucketDelayedDeliveryWithAllConsumersDisconnecting() throws Exce
Assert.assertEquals(bucketKeys, bucketKeys2);
}

@Test
public void testResetCursorClearsDelayedMessages() throws Exception {
String topic = BrokerTestUtil.newUniqueName("persistent://public/default/testResetClearsDelayed");

@Cleanup
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
.topic(topic)
.subscriptionName("sub")
.subscriptionType(SubscriptionType.Shared)
.subscribe();

@Cleanup
Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
.topic(topic)
.create();

for (int i = 0; i < 100; i++) {
producer.newMessage()
.value("msg")
.deliverAfter(1, TimeUnit.HOURS)
.send();
}

Dispatcher dispatcher = pulsar.getBrokerService().getTopicReference(topic)
.get().getSubscription("sub").getDispatcher();
Awaitility.await().untilAsserted(() ->
Assert.assertEquals(dispatcher.getNumberOfDelayedMessages(), 100));
List<String> bucketKeys =
((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties()
.keySet().stream().filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
assertFalse(bucketKeys.isEmpty());

// Resetting the cursor disconnects the consumer, so nothing gets re-tracked while we
// observe the post-reset state.
admin.topics().resetCursor(topic, "sub", MessageId.earliest);

assertEquals(dispatcher.getNumberOfDelayedMessages(), 0,

@lhotari lhotari Aug 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

[QUALITY] the new test races the client's automatic reconnect, which can re-track the delayed messages before the assertions run

These two assertions are unguarded, and the comment above the reset states the no-re-tracking assumption without enforcing it.

The reset disconnects the consumer via Consumer.disconnect (Consumer.java:496-504), which sends CommandCloseConsumer. The client reconnects after initialBackoffIntervalNanos, default 100 ms (ClientConfigurationData.java:331-334, ClientCnx.java:1284-1300). PersistentSubscription.addConsumer (:248-251) explicitly chains the reconnect behind inProgressResetCursorFuture, so the reconnect lands immediately after the reset completes — exactly where these assertions run. Once resubscribed, the dispatcher reads from the reset position and re-tracks the 100 delayed messages, restoring both the tracker count and the bucket cursor properties.

Failure scenario. On a loaded CI runner, the reset itself (bucket snapshot deletes over metadata plus the cursor ledger write) plus scheduling jitter exceeds the ~100 ms backoff that started when the consumer was disconnected at the top of the reset. The consumer resubscribes and re-tracks before this line runs, so assertEquals(dispatcher.getNumberOfDelayedMessages(), 0) sees 100, or bucketKeysAfterReset is non-empty.

Suggested fix: consumer.close() before admin.topics().resetCursor(...). PersistentSubscription.dispatcher is never nulled, so dispatcher != null still holds and clearDelayedMessages() is still exercised; dispatcher.isConsumerConnected() is then false, so disconnectFuture completes immediately and the rest of the path is unchanged.

Worth adding while you are here: the test covers neither the dispatcher == null branch nor the clear-failure branch that the description says now unfences the subscription.

"The delayed delivery tracker should be cleared by the cursor reset");
List<String> bucketKeysAfterReset =
((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties()
.keySet().stream().filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
assertTrue(bucketKeysAfterReset.isEmpty(),
"The bucket cursor properties should be removed by the cursor reset");
}

@Test
public void testIncrementPartitionsDoesNotCopyBucketDelayedDeliveryState() throws Exception {
String topic = BrokerTestUtil.newUniqueName("persistent://public/default/testBucketStatePartitionExpansion");
Expand Down
Loading