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..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; @@ -983,88 +984,106 @@ private CompletableFuture resetCursorInternal(Position finalPosition, Comp } } - disconnectFuture.whenComplete((aVoid, throwable) -> { - if (dispatcher != null) { - dispatcher.resetCloseFuture(); - } + disconnectFuture + .handle((ignore, 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; - } + if (throwable != null) { + log.error() + .exception(throwable) + .log("Failed to disconnect consumer from subscription"); - log.info() - .log("Successfully disconnected consumers from subscription, proceeding with cursor reset"); + return CompletableFuture.failedFuture( + new SubscriptionBusyException( + "Failed to disconnect consumers from subscription")); + } - 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); + log.info() + .log("Successfully disconnected consumers from subscription, proceeding with cursor reset"); + + if (dispatcher != null) { + return dispatcher.clearDelayedMessages(); } - }).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); - } - 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); + }) + .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; + }); } + 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; } 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..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; @@ -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 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()); + + consumer.close(); + + 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 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");