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 3 commits into
base: master
Choose a base branch
from
+188 −73
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
3 commits
Select commit Hold shift + click to select a range
15f944a
[fix][broker] Clear delayed delivery state before resetting the cursor
nodece Aug 25, 2026
f16e705
[fix][broker] Address review feedback on delayed-state reset clearing
nodece Sep 1, 2026
803173d
[fix][broker] Await delayed delivery clearing before the cursor reset
nodece Sep 1, 2026
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
168 changes: 95 additions & 73 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 @@ -32,6 +32,7 @@
import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
Expand All @@ -54,6 +55,7 @@
import org.apache.bookkeeper.mledger.ScanOutcome;
import org.apache.commons.lang3.tuple.MutablePair;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory;
import org.apache.pulsar.broker.intercept.BrokerInterceptor;
import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl;
import org.apache.pulsar.broker.loadbalance.extensions.data.BrokerLookupData;
Expand Down Expand Up @@ -983,88 +985,108 @@ 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;
}
disconnectFuture
.handle((ignore, throwable) -> {
if (dispatcher != null) {
dispatcher.resetCloseFuture();
}

log.info()
.log("Successfully disconnected consumers from subscription, proceeding with cursor reset");
if (throwable != null) {
log.error()
.exception(throwable)
.log("Failed to disconnect consumer from subscription");

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);
throw new CompletionException(
new SubscriptionBusyException(
"Failed to disconnect consumers from subscription"));
}
}).exceptionally(ex -> {
forceReset.completeExceptionally(ex);

log.info()
.log("Successfully disconnected consumers from subscription, proceeding with cursor reset");
return null;
});
}
})
.thenCompose(__ -> {
if (dispatcher != null) {
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.

@nodece nodece Sep 1, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Right, and it actually bit me while testing the reconnect fix in the test below. With the consumer closed before the reset, the full BucketDelayedDeliveryTest suite fails for me every single time: the property removal from the bucket delete still races asyncResetCursor persistence and the reset fails with BadVersion ("unable to persist readPosition for cursor reset"). Same pre-existing clear() gap you describe here. #26401 (still open) is where this gets fixed - picking 7f22e56 on top of this branch makes the suite pass again, verified locally. I would rather not duplicate that work in this PR, so the plan is to rebase onto #26401 once it lands.

}

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);
if (topic.isDelayedDeliveryEnabled()
&& topic.getBrokerService().getDelayedDeliveryTrackerFactory()
instanceof BucketDelayedDeliveryTrackerFactory bucketDelayedDeliveryTrackerFactory) {
return bucketDelayedDeliveryTrackerFactory.cleanResidualSnapshots(cursor);
}

@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));
}
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.

@nodece nodece Sep 1, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Fixed in f16e705. I was originally fine with the skip because re-added messages dedup through the index bitmap, but that argument only holds for reset-to-earliest: getScheduledMessages filters by the ledger GC boundary, not by the reset position, so with a reset to a later position the recovered snapshots would still hand out the skipped messages. The no-dispatcher branch now runs cleanResidualSnapshots(cursor) behind the same guard as the unsubscribe path, and testResetCursorWithoutDispatcherCleansResidualBucketSnapshots covers it (close the consumer, unload the topic, reset, assert the properties are gone).

})
.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;
});
}
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);
}

@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;
Throwable cause = FutureUtil.unwrapCompletionException(e);
future.completeExceptionally(cause instanceof BrokerServiceException exception
? exception : new BrokerServiceException(cause));
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
93 changes: 93 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 @@ -24,6 +24,7 @@
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import com.google.common.collect.Multimap;
import java.io.ByteArrayOutputStream;
Expand Down Expand Up @@ -128,6 +129,98 @@ 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());

consumer.close();

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.

@nodece nodece Sep 1, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Done in f16e705, consumer.close() before the reset. The dispatcher survives the disconnect, so the reset still clears the tracker it populated and the test still fails on master without the fix. Closing the consumer first is also what flushed out the clear() race from the comment above, so the ordering matters beyond the assertions.

"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 testResetCursorWithoutDispatcherCleansResidualBucketSnapshots() throws Exception {
String topic = BrokerTestUtil.newUniqueName("persistent://public/default/testResetNoDispatcher");

@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();
}

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

consumer.close();
admin.topics().unload(topic);

admin.topics().resetCursor(topic, "sub", MessageId.earliest);

PersistentSubscription reloadedSubscription = (PersistentSubscription) pulsar.getBrokerService()
.getTopicReference(topic).get().getSubscription("sub");
assertNull(reloadedSubscription.getDispatcher(),
"No consumer has connected since the topic was reloaded");
List<String> bucketKeysAfterReset = reloadedSubscription.getCursor().getCursorProperties()
.keySet().stream().filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
assertTrue(bucketKeysAfterReset.isEmpty(),
"A reset without a dispatcher should still remove the residual bucket cursor properties");
}

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