Skip to content

[improve][broker] Centralize mark-delete position change handling #25571

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

Closed
nodece wants to merge 1 commit into from
Closed

[improve][broker] Centralize mark-delete position change handling #25571

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
31 changes: 19 additions & 12 deletions ...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@
import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
import org.apache.pulsar.broker.service.EntryBatchSizes;
import org.apache.pulsar.broker.service.InMemoryRedeliveryTracker;
import org.apache.pulsar.broker.service.PendingAcksMap;
import org.apache.pulsar.broker.service.RedeliveryTracker;
import org.apache.pulsar.broker.service.RedeliveryTrackerDisabled;
import org.apache.pulsar.broker.service.SendMessageInfo;
Expand Down Expand Up @@ -145,7 +146,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
protected enum ReadType {
Normal, Replay
}
private Position lastMarkDeletePositionBeforeReadMoreEntries;
private volatile long readMoreEntriesCallCount;

public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor,
Expand Down Expand Up @@ -339,6 +339,24 @@ public void readMoreEntriesAsync() {
}
}

@Override
public void markDeletePositionMoveForward() {
// When the mark-delete position advances (due to ack or TTL expiry), remove stale entries that are
// now below the new mark-delete position from the redelivery tracker and each consumer's pending acks.
synchronized (PersistentDispatcherMultipleConsumers.this) {
Position mdp = cursor.getMarkDeletedPosition();
if (mdp != null) {
redeliveryMessages.removeAllUpTo(mdp.getLedgerId(), mdp.getEntryId());
for (Consumer consumer : consumerList) {
PendingAcksMap pendingAcks = consumer.getPendingAcks();
if (pendingAcks != null) {
pendingAcks.removeAllUpTo(mdp.getLedgerId(), mdp.getEntryId());
}
}
Comment on lines +349 to +355

@lhotari lhotari Apr 23, 2026 •
edited
Loading

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.

Is there a significant different in the frequency that this gets called? When this logic was previously in readMoreEntries, it would only get called there if the mark delete position moved since the last call to readMoreEntries, which is not as frequent as the mark delete position might move forward. This has been intentional behavior in the past and I have a concern that this change would cause a performance regression.

@nodece nodece Apr 23, 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.

There's likely a performance impact due to the higher invocation frequency. Thanks for pointing that out. I'll revisit the change and adjust it to avoid unnecessary executions!

}
}
}

public synchronized void readMoreEntries() {
if (cursor.isClosed()) {
log.debug("Cursor is already closed, skipping read more entries");
Expand All @@ -362,17 +380,6 @@ public synchronized void readMoreEntries() {
// increment the counter for readMoreEntries calls, to track the number of times readMoreEntries is called
readMoreEntriesCallCount++;

// remove possible expired messages from redelivery tracker and pending acks
Position markDeletePosition = cursor.getMarkDeletedPosition();
if (lastMarkDeletePositionBeforeReadMoreEntries != markDeletePosition) {
redeliveryMessages.removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
for (Consumer consumer : consumerList) {
consumer.getPendingAcks()
.removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId());
}
lastMarkDeletePositionBeforeReadMoreEntries = markDeletePosition;
}

// totalAvailablePermits may be updated by other threads
int firstAvailableConsumerPermits = getFirstAvailableConsumerPermits();
int currentTotalAvailablePermits = Math.max(totalAvailablePermits, firstAvailableConsumerPermits);
Expand Down
4 changes: 1 addition & 3 deletions ...main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@
import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo;
import org.apache.pulsar.broker.service.MessageExpirer;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.stats.Rate;
import org.jspecify.annotations.Nullable;
/**
Expand Down Expand Up @@ -211,8 +210,7 @@ public void markDeleteComplete(Object ctx) {
long numMessagesExpired = (long) ctx - cursor.getNumberOfEntriesInBacklog(false);
msgExpired.recordMultipleEvents(numMessagesExpired, 0 /* no value stats */);
totalMsgExpired.add(numMessagesExpired);
// If the subscription is a Key_Shared subscription, we should to trigger message dispatch.
if (subscription != null && subscription.getType() == SubType.Key_Shared) {
if (subscription != null && subscription.getDispatcher() != null) {
subscription.getDispatcher().markDeletePositionMoveForward();
}
expirationCheckInProgress = FALSE;
Expand Down
2 changes: 2 additions & 0 deletions ...ache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -642,6 +642,8 @@ protected int getStickyKeyHash(Entry entry) {

@Override
public void markDeletePositionMoveForward() {
// clean up stale entries from the redelivery tracker and pending acks
super.markDeletePositionMoveForward();
// reschedule a read with a backoff after moving the mark-delete position forward since there might have
// been consumers that were blocked by hash and couldn't make progress
reScheduleReadWithKeySharedUnblockingInterval();
Expand Down
37 changes: 8 additions & 29 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 @@ -515,12 +515,7 @@ public void markDeleteComplete(Object ctx) {
.attr("newMD", newMD)
.attr("oldMD", oldMD)
.log("Mark deleted messages to position from position");
// Signal the dispatchers to give chance to take extra actions
if (dispatcher != null) {
dispatcher.afterAckMessages(null, ctx);
}
// Signal the dispatchers to give chance to take extra actions
notifyTheMarkDeletePositionChanged(oldMD);
notifyTheMarkDeletePositionChanged(oldMD);
}

@Override
Expand All @@ -530,10 +525,6 @@ public void markDeleteFailed(ManagedLedgerException exception, Object ctx) {
.attr("ctx", ctx)
.exceptionMessage(exception)
.log("Failed to mark delete for position");
// Signal the dispatchers to give chance to take extra actions
if (dispatcher != null) {
dispatcher.afterAckMessages(null, ctx);
}
}
};

Expand All @@ -544,10 +535,6 @@ public void deleteComplete(Object context) {
log.debug()
.attr("context", context)
.log("Deleted message");
// Signal the dispatchers to give chance to take extra actions
if (dispatcher != null) {
dispatcher.afterAckMessages(null, context);
}
notifyTheMarkDeletePositionChanged((Position) context);
}

Expand All @@ -557,10 +544,6 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
.attr("ctx", ctx)
.exceptionMessage(exception)
.log("Failed to delete message");
// Signal the dispatchers to give chance to take extra actions
if (dispatcher != null) {
dispatcher.afterAckMessages(exception, ctx);
}
}
};

Expand All @@ -579,6 +562,7 @@ private void notifyTheMarkDeletePositionChanged(Position oldPosition) {
handleReplicatedSubscriptionsUpdate(newMD);

if (dispatcher != null) {
dispatcher.afterAckMessages(null, null);
dispatcher.markDeletePositionMoveForward();
}
}
Expand Down Expand Up @@ -784,6 +768,7 @@ public CompletableFuture<Void> clearBacklog() {
.attr("entriesInBacklog", cursor.getNumberOfEntriesInBacklog(false))
.log("Backlog size before clearing");

final Position oldMarkDeletePosition = cursor.getMarkDeletedPosition();
cursor.asyncClearBacklog(new ClearBacklogCallback() {
@Override
public void clearBacklogComplete(Object ctx) {
Expand All @@ -798,10 +783,10 @@ public void clearBacklogComplete(Object ctx) {
future.complete(null);
}
});
dispatcher.afterAckMessages(null, ctx);
} else {
future.complete(null);
}
notifyTheMarkDeletePositionChanged(oldMarkDeletePosition);
}

@Override
Expand All @@ -810,9 +795,6 @@ public void clearBacklogFailed(ManagedLedgerException exception, Object ctx) {
.exception(exception)
.log("Failed to clear backlog");
future.completeExceptionally(exception);
if (dispatcher != null) {
dispatcher.afterAckMessages(exception, ctx);
}
}
}, null);

Expand All @@ -827,6 +809,7 @@ public CompletableFuture<Void> skipMessages(int numMessagesToSkip) {
.attr("numMessagesToSkip", numMessagesToSkip)
.attr("entriesInBacklog", cursor.getNumberOfEntriesInBacklog(false))
.log("Skipping messages");
final Position oldMarkDeletePosition = cursor.getMarkDeletedPosition();
cursor.asyncSkipEntries(numMessagesToSkip, IndividualDeletedEntries.Exclude,
new AsyncCallbacks.SkipEntriesCallback() {
@Override
Expand All @@ -836,9 +819,7 @@ public void skipEntriesComplete(Object ctx) {
.attr("entriesInBacklog", cursor.getNumberOfEntriesInBacklog(false))
.log("Skipped messages");
future.complete(null);
if (dispatcher != null) {
dispatcher.afterAckMessages(null, ctx);
}
notifyTheMarkDeletePositionChanged(oldMarkDeletePosition);
}

@Override
Expand All @@ -848,9 +829,6 @@ public void skipEntriesFailed(ManagedLedgerException exception, Object ctx) {
.exception(exception)
.log("Failed to skip messages");
future.completeExceptionally(exception);
if (dispatcher != null) {
dispatcher.afterAckMessages(exception, ctx);
}
}
}, null);

Expand Down Expand Up @@ -983,6 +961,7 @@ private CompletableFuture<Void> resetCursorInternal(Position finalPosition, Comp
}

forceReset.thenAccept(forceResetValue -> {
final Position oldMarkDeletePosition = cursor.getMarkDeletedPosition();
cursor.asyncResetCursor(finalPosition, forceResetValue, new AsyncCallbacks.ResetCursorCallback() {
@Override
public void resetComplete(Object ctx) {
Expand All @@ -991,8 +970,8 @@ public void resetComplete(Object ctx) {
.log("Successfully reset subscription to position");
if (dispatcher != null) {
dispatcher.cursorIsReset();
dispatcher.afterAckMessages(null, finalPosition);
}
notifyTheMarkDeletePositionChanged(oldMarkDeletePosition);
IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
inProgressResetCursorFuture = null;
future.complete(null);
Expand Down
Loading