From 15f944a96ff856d98c1b7e7a68d05b006c5a5fc3 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Tue, 25 Aug 2026 18:02:16 +0800 Subject: [PATCH 1/3] [fix][broker] Clear delayed delivery state before resetting the cursor resetCursorInternal() reset the cursor without touching the delayed delivery tracker, although a reset moves the consumption baseline the tracker state was derived from: bucket snapshots, index bits and queued entries from before the reset survived into the replay, and in-flight trims/loads/deletes could race the replayed state. Clear the delayed messages and wait for the clear to settle after disconnecting consumers and before asyncResetCursor, with the whole reset flow chained into a single error path. Without a dispatcher there is no in-flight work and the recovered bucket snapshots stay valid for the replay, so there is nothing to clear. --- .../persistent/PersistentSubscription.java | 147 +++++++++--------- .../persistent/BucketDelayedDeliveryTest.java | 45 ++++++ 2 files changed, 116 insertions(+), 76 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index 9ba3a4be50e28..d64825c1b0277 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -983,88 +983,83 @@ private CompletableFuture 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 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(__ -> { + log.info() + .log("Successfully disconnected consumers from subscription, proceeding with cursor reset"); + if (dispatcher != null) { + dispatcher.resetCloseFuture(); + return dispatcher.clearDelayedMessages(); + } + return CompletableFuture.completedFuture(null); + }) + .thenCompose(__ -> { + CompletableFuture 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; } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java index 84020f6aea778..5cd3986421eb8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java @@ -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 consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName("sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + + @Cleanup + Producer 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 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, + "The delayed delivery tracker should be cleared by the cursor reset"); + List 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"); From f16e705c12097ebe9df8d09e6a581e92fd435d56 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Tue, 1 Sep 2026 11:47:57 +0800 Subject: [PATCH 2/3] [fix][broker] Address review feedback on delayed-state reset clearing - Restore the disconnect-failure branch dropped by the thenCompose rewrite: reset the dispatcher close future and fail with SubscriptionBusyException (HTTP 412) instead of a generic BrokerServiceException, and unwrap CompletionException so typed exceptions pass through unwrapped. - Clean residual bucket snapshots when no dispatcher exists, mirroring the unsubscribe path, so a later consumer reconnect cannot recover pre-reset buckets from the cursor properties. - Close the consumer before resetting in the regression test so the client's automatic reconnect cannot re-track the replayed delayed messages before the assertions, and add a test for the no-dispatcher cleanup path. Assisted-by: Claude Code --- .../persistent/PersistentSubscription.java | 30 +++++++++-- .../persistent/BucketDelayedDeliveryTest.java | 52 ++++++++++++++++++- 2 files changed, 77 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index d64825c1b0277..4975006dbd1a1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -54,6 +54,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; @@ -984,13 +985,34 @@ private CompletableFuture resetCursorInternal(Position finalPosition, Comp } disconnectFuture - .thenCompose(__ -> { + .handle((ignore, throwable) -> { + if (dispatcher != null) { + dispatcher.resetCloseFuture(); + } + + if (throwable != null) { + log.error() + .exception(throwable) + .log("Failed to disconnect consumer from subscription"); + + return CompletableFuture.failedFuture( + new SubscriptionBusyException( + "Failed to disconnect consumers from subscription")); + } + log.info() .log("Successfully disconnected consumers from subscription, proceeding with cursor reset"); + if (dispatcher != null) { - dispatcher.resetCloseFuture(); return dispatcher.clearDelayedMessages(); } + + if (topic.isDelayedDeliveryEnabled() + && topic.getBrokerService().getDelayedDeliveryTrackerFactory() + instanceof BucketDelayedDeliveryTrackerFactory bucketDelayedDeliveryTrackerFactory) { + return bucketDelayedDeliveryTrackerFactory.cleanResidualSnapshots(cursor); + } + return CompletableFuture.completedFuture(null); }) .thenCompose(__ -> { @@ -1057,7 +1079,9 @@ public void resetFailed(ManagedLedgerException exception, Object ctx) { .log("Error while resetting cursor"); IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE); inProgressResetCursorFuture = null; - future.completeExceptionally(new BrokerServiceException(e)); + Throwable cause = FutureUtil.unwrapCompletionException(e); + future.completeExceptionally(cause instanceof BrokerServiceException exception + ? exception : new BrokerServiceException(cause)); return null; }); return future; diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java index 5cd3986421eb8..fa83482d439ac 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java @@ -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; @@ -160,8 +161,8 @@ public void testResetCursorClearsDelayedMessages() throws Exception { .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. + consumer.close(); + admin.topics().resetCursor(topic, "sub", MessageId.earliest); assertEquals(dispatcher.getNumberOfDelayedMessages(), 0, @@ -173,6 +174,53 @@ public void testResetCursorClearsDelayedMessages() throws Exception { "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 consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName("sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + + @Cleanup + Producer 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 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 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"); From 803173d9bd0a87778d576bc7c2be231147286798 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Tue, 1 Sep 2026 16:40:03 +0800 Subject: [PATCH 3/3] [fix][broker] Await delayed delivery clearing before the cursor reset The review-feedback rewrite switched the disconnect stage from thenCompose to handle(), but handle() does not flatten a returned CompletableFuture: the delayed-delivery clear (and the residual bucket snapshot cleanup without a dispatcher) was never awaited, so asyncResetCursor could race the bucket cursor-property writes from the clear and fail the cursor metadata update with a BadVersion conflict, surfacing as HTTP 412 "unable to persist readPosition for cursor reset". Fail the disconnect stage by throwing a wrapped SubscriptionBusyException and move the clearing into a following thenCompose so the reset only starts after the clear settles; the disconnect failure still maps to SubscriptionBusyException via the unwrapping in the final error path. Assisted-by: Claude Code --- .../broker/service/persistent/PersistentSubscription.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index 4975006dbd1a1..033434ea5ac01 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -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; @@ -995,14 +996,16 @@ private CompletableFuture resetCursorInternal(Position finalPosition, Comp .exception(throwable) .log("Failed to disconnect consumer from subscription"); - return CompletableFuture.failedFuture( + throw new CompletionException( new SubscriptionBusyException( "Failed to disconnect consumers from subscription")); } log.info() .log("Successfully disconnected consumers from subscription, proceeding with cursor reset"); - + return null; + }) + .thenCompose(__ -> { if (dispatcher != null) { return dispatcher.clearDelayedMessages(); }