Skip to content

Commit f7e0723

Browse files
nodecenodece
committed
[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. 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.
1 parent 9890872 commit f7e0723

2 files changed

Lines changed: 122 additions & 55 deletions

File tree

‎pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java‎

Lines changed: 77 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -1002,67 +1002,89 @@ private CompletableFuture<Void> resetCursorInternal(Position finalPosition, Comp
10021002
log.info()
10031003
.log("Successfully disconnected consumers from subscription, proceeding with cursor reset");
10041004

1005-
CompletableFuture<Boolean> forceReset = new CompletableFuture<>();
1006-
if (topic.getTopicCompactionService() == null) {
1007-
forceReset.complete(false);
1008-
} else {
1009-
topic.getTopicCompactionService().getLastCompactedPosition().thenAccept(lastCompactedPosition -> {
1010-
Position resetTo = finalPosition;
1011-
if (lastCompactedPosition != null && resetTo.compareTo(lastCompactedPosition.getLedgerId(),
1012-
lastCompactedPosition.getEntryId()) <= 0) {
1013-
forceReset.complete(true);
1014-
} else {
1015-
forceReset.complete(false);
1016-
}
1017-
}).exceptionally(ex -> {
1018-
forceReset.completeExceptionally(ex);
1019-
return null;
1020-
});
1005+
CompletableFuture<Void> clearDelayedMessagesFuture;
1006+
try {
1007+
clearDelayedMessagesFuture = dispatcher != null
1008+
? dispatcher.clearDelayedMessages()
1009+
: CompletableFuture.completedFuture(null);
1010+
} catch (Throwable t) {
1011+
clearDelayedMessagesFuture = FutureUtil.failedFuture(t);
10211012
}
10221013

1023-
forceReset.thenAccept(forceResetValue -> {
1024-
cursor.asyncResetCursor(finalPosition, forceResetValue, new AsyncCallbacks.ResetCursorCallback() {
1025-
@Override
1026-
public void resetComplete(Object ctx) {
1027-
log.debug()
1028-
.attr("finalPosition", finalPosition)
1029-
.log("Successfully reset subscription to position");
1030-
if (dispatcher != null) {
1031-
dispatcher.cursorIsReset();
1032-
dispatcher.afterAckMessages(null, finalPosition);
1033-
}
1034-
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
1035-
inProgressResetCursorFuture = null;
1036-
future.complete(null);
1037-
}
1014+
clearDelayedMessagesFuture.whenComplete((__, clearEx) -> {
1015+
if (clearEx != null) {
1016+
log.error()
1017+
.exception(clearEx)
1018+
.log("Error while clearing delayed messages during cursor reset");
1019+
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
1020+
inProgressResetCursorFuture = null;
1021+
future.completeExceptionally(new BrokerServiceException(clearEx));
1022+
return;
1023+
}
10381024

1039-
@Override
1040-
public void resetFailed(ManagedLedgerException exception, Object ctx) {
1041-
log.error()
1042-
.attr("finalPosition", finalPosition)
1043-
.exception(exception)
1044-
.log("Failed to reset subscription to position");
1045-
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
1046-
inProgressResetCursorFuture = null;
1047-
// todo - retry on InvalidCursorPositionException
1048-
// or should we just ask user to retry one more time?
1049-
if (exception instanceof InvalidCursorPositionException) {
1050-
future.completeExceptionally(new SubscriptionInvalidCursorPosition(exception.getMessage()));
1051-
} else if (exception instanceof ConcurrentFindCursorPositionException) {
1052-
future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
1025+
CompletableFuture<Boolean> forceReset = new CompletableFuture<>();
1026+
if (topic.getTopicCompactionService() == null) {
1027+
forceReset.complete(false);
1028+
} else {
1029+
topic.getTopicCompactionService().getLastCompactedPosition().thenAccept(lastCompactedPosition -> {
1030+
Position resetTo = finalPosition;
1031+
if (lastCompactedPosition != null && resetTo.compareTo(lastCompactedPosition.getLedgerId(),
1032+
lastCompactedPosition.getEntryId()) <= 0) {
1033+
forceReset.complete(true);
10531034
} else {
1054-
future.completeExceptionally(new BrokerServiceException(exception));
1035+
forceReset.complete(false);
10551036
}
1056-
}
1037+
}).exceptionally(ex -> {
1038+
forceReset.completeExceptionally(ex);
1039+
return null;
1040+
});
1041+
}
1042+
1043+
forceReset.thenAccept(forceResetValue -> {
1044+
cursor.asyncResetCursor(finalPosition, forceResetValue, new AsyncCallbacks.ResetCursorCallback() {
1045+
@Override
1046+
public void resetComplete(Object ctx) {
1047+
log.debug()
1048+
.attr("finalPosition", finalPosition)
1049+
.log("Successfully reset subscription to position");
1050+
if (dispatcher != null) {
1051+
dispatcher.cursorIsReset();
1052+
dispatcher.afterAckMessages(null, finalPosition);
1053+
}
1054+
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
1055+
inProgressResetCursorFuture = null;
1056+
future.complete(null);
1057+
}
1058+
1059+
@Override
1060+
public void resetFailed(ManagedLedgerException exception, Object ctx) {
1061+
log.error()
1062+
.attr("finalPosition", finalPosition)
1063+
.exception(exception)
1064+
.log("Failed to reset subscription to position");
1065+
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
1066+
inProgressResetCursorFuture = null;
1067+
// todo - retry on InvalidCursorPositionException
1068+
// or should we just ask user to retry one more time?
1069+
if (exception instanceof InvalidCursorPositionException) {
1070+
future.completeExceptionally(
1071+
new SubscriptionInvalidCursorPosition(exception.getMessage()));
1072+
} else if (exception instanceof ConcurrentFindCursorPositionException) {
1073+
future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
1074+
} else {
1075+
future.completeExceptionally(new BrokerServiceException(exception));
1076+
}
1077+
}
1078+
});
1079+
}).exceptionally((e) -> {
1080+
log.error()
1081+
.exception(e)
1082+
.log("Error while resetting cursor");
1083+
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
1084+
inProgressResetCursorFuture = null;
1085+
future.completeExceptionally(new BrokerServiceException(e));
1086+
return null;
10571087
});
1058-
}).exceptionally((e) -> {
1059-
log.error()
1060-
.exception(e)
1061-
.log("Error while resetting cursor");
1062-
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
1063-
inProgressResetCursorFuture = null;
1064-
future.completeExceptionally(new BrokerServiceException(e));
1065-
return null;
10661088
});
10671089
});
10681090
return future;

‎pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,51 @@ public void testBucketDelayedDeliveryWithAllConsumersDisconnecting() throws Exce
128128
Assert.assertEquals(bucketKeys, bucketKeys2);
129129
}
130130

131+
@Test
132+
public void testResetCursorClearsDelayedMessages() throws Exception {
133+
String topic = BrokerTestUtil.newUniqueName("persistent://public/default/testResetClearsDelayed");
134+
135+
@Cleanup
136+
Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
137+
.topic(topic)
138+
.subscriptionName("sub")
139+
.subscriptionType(SubscriptionType.Shared)
140+
.subscribe();
141+
142+
@Cleanup
143+
Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
144+
.topic(topic)
145+
.create();
146+
147+
for (int i = 0; i < 100; i++) {
148+
producer.newMessage()
149+
.value("msg")
150+
.deliverAfter(1, TimeUnit.HOURS)
151+
.send();
152+
}
153+
154+
Dispatcher dispatcher = pulsar.getBrokerService().getTopicReference(topic)
155+
.get().getSubscription("sub").getDispatcher();
156+
Awaitility.await().untilAsserted(() ->
157+
Assert.assertEquals(dispatcher.getNumberOfDelayedMessages(), 100));
158+
List<String> bucketKeys =
159+
((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties()
160+
.keySet().stream().filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
161+
assertFalse(bucketKeys.isEmpty());
162+
163+
// Resetting the cursor disconnects the consumer, so nothing gets re-tracked while we
164+
// observe the post-reset state.
165+
admin.topics().resetCursor(topic, "sub", MessageId.earliest);
166+
167+
assertEquals(dispatcher.getNumberOfDelayedMessages(), 0,
168+
"The delayed delivery tracker should be cleared by the cursor reset");
169+
List<String> bucketKeysAfterReset =
170+
((AbstractPersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties()
171+
.keySet().stream().filter(x -> x.startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX)).toList();
172+
assertTrue(bucketKeysAfterReset.isEmpty(),
173+
"The bucket cursor properties should be removed by the cursor reset");
174+
}
175+
131176
@Test
132177
public void testIncrementPartitionsDoesNotCopyBucketDelayedDeliveryState() throws Exception {
133178
String topic = BrokerTestUtil.newUniqueName("persistent://public/default/testBucketStatePartitionExpansion");

0 commit comments

Comments
 (0)