From 79241e2d9dd103b776783a0bcc685e1002e0a44b Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Tue, 28 Apr 2026 20:06:37 +0800 Subject: [PATCH 1/5] [fix][broker] Move pending acks cleanup to selected mark-delete callbacks --- .../pulsar/broker/service/Consumer.java | 57 +++++++++---------- .../pulsar/broker/service/Dispatcher.java | 17 ++++++ .../pulsar/broker/service/PendingAcksMap.java | 46 ++++++++++++--- ...PersistentDispatcherMultipleConsumers.java | 20 +++---- .../PersistentMessageExpiryMonitor.java | 16 +++++- .../persistent/PersistentSubscription.java | 8 +++ 6 files changed, 111 insertions(+), 53 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 3a0bf99eb6d63..410388560d922 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -385,9 +385,18 @@ public Future sendMessages(final List entries, } else { stickyKeyHash = stickyKeyHashes.get(i); } - boolean sendingAllowed = - pendingAcks.addPendingAckIfAllowed(entry.getLedgerId(), entry.getEntryId(), batchSize, - stickyKeyHash); + boolean sendingAllowed; + long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i); + int remainingUnacked; + if (ackSet != null) { + remainingUnacked = BitSet.valueOf(ackSet).cardinality(); + unackedMessages -= (batchSize - remainingUnacked); + } else { + remainingUnacked = batchSize; + } + sendingAllowed = + pendingAcks.addPendingAckIfAllowed(entry.getLedgerId(), entry.getEntryId(), + remainingUnacked, stickyKeyHash); if (!sendingAllowed) { // sending isn't allowed when pending acks doesn't accept adding the entry // this happens when Key_Shared draining hashes contains the stickyKeyHash @@ -401,10 +410,6 @@ public Future sendMessages(final List entries, .attr("batchSize", batchSize) .log("Skipping sending of entry since adding to pending acks failed"); } else { - long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i); - if (ackSet != null) { - unackedMessages -= (batchSize - BitSet.valueOf(ackSet).cardinality()); - } log.debug() .attr("ledgerId", entry.getLedgerId()) .attr("entryId", entry.getEntryId()) @@ -596,6 +601,10 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map 0) { + ackOwnerConsumer.getPendingAcks().updateRemainingUnacked( + position.getLedgerId(), position.getEntryId(), (int) ackedCount); + } addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); } else { position = PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId()); @@ -679,6 +688,12 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { } AckSetStateUtil.getAckSetState(position).setAckSet(ackSets); ackedCount = getAckedCountForTransactionAck(batchSize, ackSets); + // Update the stored remaining unacked count for this entry so that + // removal paths (expire, skip, redeliver) can use it directly. + if (ackedCount > 0) { + ackOwnerConsumer.getPendingAcks().updateRemainingUnacked( + position.getLedgerId(), position.getEntryId(), (int) ackedCount); + } } addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); @@ -747,19 +762,6 @@ private long getAckedCountForTransactionAck(int batchSize, long[] ackSets) { return ackedCount; } - private long getUnAckedCountForBatchIndexLevelEnabled(Position position, int batchSize) { - long unAckedCount = batchSize; - if (isAcknowledgmentAtBatchIndexLevelEnabled) { - long[] cursorAckSet = getCursorAckSet(position); - if (cursorAckSet != null) { - BitSetRecyclable cursorBitSet = BitSetRecyclable.create().resetWords(cursorAckSet); - unAckedCount = cursorBitSet.cardinality(); - cursorBitSet.recycle(); - } - } - return unAckedCount; - } - private void checkAckValidationError(CommandAck ack, Position position) { if (ack.hasValidationError()) { log.warn() @@ -1139,9 +1141,8 @@ public void removePendingAcksUpToPositionAndDecrementUnacked(long markDeleteLedg MutableInt mutableTotalUnacked = new MutableInt(0); pendingAcks.removeAllUpTo(markDeleteLedgerId, markDeleteEntryId, - (ledgerId, entryId, batchSize, stickyKeyHash) -> { - mutableTotalUnacked.add((int) getUnAckedCountForBatchIndexLevelEnabled( - PositionFactory.create(ledgerId, entryId), batchSize)); + (ledgerId, entryId, remainingUnacked, stickyKeyHash) -> { + mutableTotalUnacked.add(remainingUnacked); }); int totalUnacked = mutableTotalUnacked.intValue(); if (totalUnacked > 0) { @@ -1160,11 +1161,8 @@ public void redeliverUnacknowledgedMessages(long consumerEpoch) { if (pendingAcks != null) { List pendingPositions = new ArrayList<>((int) pendingAcks.size()); MutableInt totalRedeliveryMessages = new MutableInt(0); - pendingAcks.forEachAndClear((ledgerId, entryId, batchSize, stickyKeyHash) -> { - int unAckedCount = - (int) getUnAckedCountForBatchIndexLevelEnabled(PositionFactory.create(ledgerId, entryId), - batchSize); - totalRedeliveryMessages.add(unAckedCount); + pendingAcks.forEachAndClear((ledgerId, entryId, remainingUnacked, stickyKeyHash) -> { + totalRedeliveryMessages.add(remainingUnacked); pendingPositions.add(PositionFactory.create(ledgerId, entryId)); }); @@ -1193,8 +1191,7 @@ public void redeliverUnacknowledgedMessages(List messageIds) { Position position = PositionFactory.create(msg.getLedgerId(), msg.getEntryId()); IntIntPair pendingAck = pendingAcks.removeAndGet(position.getLedgerId(), position.getEntryId()); if (pendingAck != null) { - int unAckedCount = (int) getUnAckedCountForBatchIndexLevelEnabled(position, pendingAck.leftInt()); - totalRedeliveryMessages += unAckedCount; + totalRedeliveryMessages += pendingAck.leftInt(); pendingPositions.add(position); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java index e19deb34e31b9..c6db0a65ae3ea 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java @@ -142,6 +142,23 @@ default boolean checkAndUnblockIfStuck() { return false; } + /** + * Prune pending-ack entries up to the specified position and updates the + * consumer unacked-message counters accordingly. + * + *

This hook is invoked after the cursor mark-delete operation completes + * (for example, during message expiry, skip, or clear backlog). Since the + * cursor ack set may no longer be available after mark-delete, the counter + * adjustment relies on the remaining unacked count stored in the + * {@code PendingAcksMap} entries. + * + * @param ledgerId the ledger ID of the inclusive upper bound position + * @param entryId the entry ID of the inclusive upper bound position + */ + default void prunePendingAcksUpToPosition(long ledgerId, long entryId) { + // No-op by default + } + /** * A callback hook after acknowledge messages. * @param exOfDeletion the ex of {@link org.apache.bookkeeper.mledger.ManagedCursor#asyncDelete}, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java index 8be69aa7879e6..70f1a7dd247c8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PendingAcksMap.java @@ -87,12 +87,13 @@ public interface PendingAcksConsumer { /** * Accept a pending acknowledgment. * - * @param ledgerId the ledger ID - * @param entryId the entry ID - * @param batchSize the batch size - * @param stickyKeyHash the sticky key hash + * @param ledgerId the ledger ID + * @param entryId the entry ID + * @param remainingUnacked the number of remaining unacked messages in this entry + * (accounts for batch index level acknowledgments) + * @param stickyKeyHash the sticky key hash */ - void accept(long ledgerId, long entryId, int batchSize, int stickyKeyHash); + void accept(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash); } private final Consumer consumer; @@ -122,11 +123,12 @@ public interface PendingAcksConsumer { * * @param ledgerId the ledger ID * @param entryId the entry ID - * @param batchSize the batch size + * @param remainingUnacked the number of remaining unacked messages in this entry + * (for batch entries with some indexes already acked, this may be less than batchSize) * @param stickyKeyHash the sticky key hash * @return true if the pending ack was added, and it's allowed to send a message, false otherwise */ - public boolean addPendingAckIfAllowed(long ledgerId, long entryId, int batchSize, int stickyKeyHash) { + public boolean addPendingAckIfAllowed(long ledgerId, long entryId, int remainingUnacked, int stickyKeyHash) { try { writeLock.lock(); // prevent adding sticky hash to pending acks if the PendingAcksMap has already been closed @@ -143,7 +145,7 @@ public boolean addPendingAckIfAllowed(long ledgerId, long entryId, int batchSize } TreeMap ledgerPendingAcks = pendingAcks.computeIfAbsent(ledgerId, k -> new TreeMap<>()); - ledgerPendingAcks.put(entryId, IntIntPair.of(batchSize, stickyKeyHash)); + ledgerPendingAcks.put(entryId, IntIntPair.of(remainingUnacked, stickyKeyHash)); return true; } finally { writeLock.unlock(); @@ -311,6 +313,34 @@ public boolean remove(long ledgerId, long entryId, int batchSize, int stickyKeyH } } + /** + * Atomically update the remaining unacked count for a pending ack entry by subtracting the given delta. + * Called from the ack handler after computing the number of batch indexes acknowledged in a partial ack. + * + * @param ledgerId the ledger ID + * @param entryId the entry ID + * @param ackedDelta the number of batch indexes that were just acknowledged + * @return true if the entry was found and updated, false otherwise + */ + public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta) { + try { + writeLock.lock(); + TreeMap ledgerMap = pendingAcks.get(ledgerId); + if (ledgerMap == null) { + return false; + } + IntIntPair current = ledgerMap.get(entryId); + if (current == null) { + return false; + } + int newRemaining = current.leftInt() - ackedDelta; + ledgerMap.put(entryId, IntIntPair.of(newRemaining, current.rightInt())); + return true; + } finally { + writeLock.unlock(); + } + } + /** * Remove the pending ack for the given ledger ID and entry ID. * diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index 65ed8a5acc306..ceef012035eda 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java @@ -145,7 +145,6 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis protected enum ReadType { Normal, Replay } - private Position lastMarkDeletePositionBeforeReadMoreEntries; private volatile long readMoreEntriesCallCount; public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCursor cursor, @@ -362,17 +361,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.removePendingAcksUpToPositionAndDecrementUnacked( - markDeletePosition.getLedgerId(), markDeletePosition.getEntryId()); - } - lastMarkDeletePositionBeforeReadMoreEntries = markDeletePosition; - } - // totalAvailablePermits may be updated by other threads int firstAvailableConsumerPermits = getFirstAvailableConsumerPermits(); int currentTotalAvailablePermits = Math.max(totalAvailablePermits, firstAvailableConsumerPermits); @@ -600,6 +588,14 @@ public CopyOnWriteArrayList getConsumers() { return consumerList; } + @Override + public void prunePendingAcksUpToPosition(long ledgerId, long entryId) { + redeliveryMessages.removeAllUpTo(ledgerId, entryId); + for (Consumer consumer : consumerList) { + consumer.removePendingAcksUpToPositionAndDecrementUnacked(ledgerId, entryId); + } + } + @Override public synchronized boolean canUnsubscribe(Consumer consumer) { return consumerList.size() == 1 && consumerSet.contains(consumer); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java index bfc8951be4960..bf1c6ac4dc9c4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java @@ -38,6 +38,7 @@ import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl; import org.apache.bookkeeper.mledger.proto.ManagedLedgerInfo; +import org.apache.pulsar.broker.service.Dispatcher; import org.apache.pulsar.broker.service.MessageExpirer; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; @@ -211,9 +212,18 @@ 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) { - subscription.getDispatcher().markDeletePositionMoveForward(); + if (subscription != null) { + Dispatcher dispatcher = subscription.getDispatcher(); + if (dispatcher != null) { + Position mdPos = cursor.getMarkDeletedPosition(); + if (mdPos != null) { + dispatcher.prunePendingAcksUpToPosition(mdPos.getLedgerId(), mdPos.getEntryId()); + } + // If the subscription is a Key_Shared subscription, we should to trigger message dispatch. + if (subscription.getType() == SubType.Key_Shared) { + dispatcher.markDeletePositionMoveForward(); + } + } } expirationCheckInProgress = FALSE; log.debug() 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 cb60a70a0c32e..8ce5ad5c897cb 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 @@ -798,6 +798,10 @@ public void clearBacklogComplete(Object ctx) { future.complete(null); } }); + Position mdPos = cursor.getMarkDeletedPosition(); + if (mdPos != null) { + dispatcher.prunePendingAcksUpToPosition(mdPos.getLedgerId(), mdPos.getEntryId()); + } dispatcher.afterAckMessages(null, ctx); } else { future.complete(null); @@ -837,6 +841,10 @@ public void skipEntriesComplete(Object ctx) { .log("Skipped messages"); future.complete(null); if (dispatcher != null) { + Position mdPos = cursor.getMarkDeletedPosition(); + if (mdPos != null) { + dispatcher.prunePendingAcksUpToPosition(mdPos.getLedgerId(), mdPos.getEntryId()); + } dispatcher.afterAckMessages(null, ctx); } } From 17710af2ac2c6cd65d9a1d4d0885dd89eb201491 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Wed, 29 Apr 2026 12:18:58 +0800 Subject: [PATCH 2/5] Address comment --- .../pulsar/broker/service/Consumer.java | 82 ++++++++++++------- .../pulsar/broker/service/Dispatcher.java | 26 ++---- ...PersistentDispatcherMultipleConsumers.java | 12 ++- .../PersistentMessageExpiryMonitor.java | 10 +-- ...tStickyKeyDispatcherMultipleConsumers.java | 1 + ...KeyDispatcherMultipleConsumersClassic.java | 1 + .../persistent/PersistentSubscription.java | 14 +--- 7 files changed, 74 insertions(+), 72 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 410388560d922..ff489ade39729 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -585,7 +585,7 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map ackOwnerConsumerAndBatchSize = getAckOwnerConsumerAndBatchSize(msgId.getLedgerId(), msgId.getEntryId()); Consumer ackOwnerConsumer = ackOwnerConsumerAndBatchSize.left(); - long ackedCount; + long ackedCount = 0; int batchSize = ackOwnerConsumerAndBatchSize.rightInt(); if (msgId.getAckSetsCount() > 0) { long[] ackSets = new long[msgId.getAckSetsCount()]; @@ -602,14 +602,19 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map 0) { - ackOwnerConsumer.getPendingAcks().updateRemainingUnacked( + boolean updated = ackOwnerConsumer.updateRemainingUnacked( position.getLedgerId(), position.getEntryId(), (int) ackedCount); + if (updated) { + addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); + } } - addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); } else { position = PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId()); - ackedCount = getAckedCountForMsgIdNoAckSets(batchSize, position, ackOwnerConsumer); - if (checkCanRemovePendingAcksAndHandle(ackOwnerConsumer, position, msgId)) { + // Atomically remove the entry and get the stored remainingUnacked. + IntIntPair removed = ackOwnerConsumer.removePendingAckAndGet( + position.getLedgerId(), position.getEntryId()); + if (removed != null) { + ackedCount = removed.leftInt(); addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); updateBlockedConsumerOnUnackedMsgs(ackOwnerConsumer); } @@ -688,18 +693,22 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { } AckSetStateUtil.getAckSetState(position).setAckSet(ackSets); ackedCount = getAckedCountForTransactionAck(batchSize, ackSets); - // Update the stored remaining unacked count for this entry so that - // removal paths (expire, skip, redeliver) can use it directly. if (ackedCount > 0) { - ackOwnerConsumer.getPendingAcks().updateRemainingUnacked( + boolean updated = ackOwnerConsumer.updateRemainingUnacked( position.getLedgerId(), position.getEntryId(), (int) ackedCount); + if (updated) { + addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); + } + } + } else { + IntIntPair removed = ackOwnerConsumer.removePendingAckAndGet( + position.getLedgerId(), position.getEntryId()); + if (removed != null) { + addAndGetUnAckedMsgs(ackOwnerConsumer, -removed.leftInt()); + updateBlockedConsumerOnUnackedMsgs(ackOwnerConsumer); } } - addAndGetUnAckedMsgs(ackOwnerConsumer, -(int) ackedCount); - - checkCanRemovePendingAcksAndHandle(ackOwnerConsumer, position, msgId); - checkAckValidationError(ack, position); totalAckCount.add(ackedCount); @@ -723,16 +732,6 @@ private CompletableFuture individualAckWithTransaction(CommandAck ack) { return completableFuture.thenApply(__ -> totalAckCount.sum()); } - private long getAckedCountForMsgIdNoAckSets(int batchSize, Position position, Consumer consumer) { - if (isAcknowledgmentAtBatchIndexLevelEnabled && Subscription.isIndividualAckMode(subType)) { - long[] cursorAckSet = getCursorAckSet(position); - if (cursorAckSet != null) { - return getAckedCountForBatchIndexLevelEnabled(position, batchSize, EMPTY_ACK_SET, consumer); - } - } - return batchSize; - } - private long getAckedCountForBatchIndexLevelEnabled(Position position, int batchSize, long[] ackSets, Consumer consumer) { long ackedCount = 0; @@ -771,14 +770,6 @@ private void checkAckValidationError(CommandAck ack, Position position) { } } - private boolean checkCanRemovePendingAcksAndHandle(Consumer ackOwnedConsumer, - Position position, MessageIdData msgId) { - if (Subscription.isIndividualAckMode(subType) && msgId.getAckSetsCount() == 0) { - return removePendingAcks(ackOwnedConsumer, position); - } - return false; - } - /** * Retrieves the acknowledgment owner consumer and batch size for the specified ledgerId and entryId. * @@ -1123,6 +1114,37 @@ public PendingAcksMap getPendingAcks() { return pendingAcks; } + /** + * Atomically decrement the remaining unacked count for the specified position + * by the given acknowledged delta. + * + *

No-op if {@code pendingAcks} is not initialized. + * + * @return {@code true} if the update succeeds or pendingAcks is null; + * {@code false} otherwise + */ + public boolean updateRemainingUnacked(long ledgerId, long entryId, int ackedDelta) { + if (pendingAcks != null) { + return pendingAcks.updateRemainingUnacked(ledgerId, entryId, ackedDelta); + } + return true; + } + + /** + * Atomically remove the pending ack entry and return its stored values. + * + *

No-op if {@code pendingAcks} is not initialized. + * + * @return the removed {@link IntIntPair#leftInt() remainingUnacked} and + * {@link IntIntPair#rightInt() stickyKeyHash}, or {@code null} if not found + */ + public IntIntPair removePendingAckAndGet(long ledgerId, long entryId) { + if (pendingAcks != null) { + return pendingAcks.removeAndGet(ledgerId, entryId); + } + return null; + } + /** * Remove all pending acks up to the given mark-delete position and decrement the consumer's unacked message * counter by the remaining unacked count for each removed entry. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java index c6db0a65ae3ea..9f59d4cd175b9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java @@ -131,6 +131,15 @@ default void cursorIsReset() { //No-op } + /** + * This hook is invoked after cursor mark-delete operations triggered by + * message removal flows such as expiry, skip, or clear backlog, but not for + * regular ack-driven mark-delete operations due to their higher frequency. + * + *

Since the cursor ack set may no longer be available after mark-delete, + * the cleanup logic relies on the remaining unacked count stored in + * {@code PendingAcksMap} entries. + */ default void markDeletePositionMoveForward() { // No-op } @@ -142,23 +151,6 @@ default boolean checkAndUnblockIfStuck() { return false; } - /** - * Prune pending-ack entries up to the specified position and updates the - * consumer unacked-message counters accordingly. - * - *

This hook is invoked after the cursor mark-delete operation completes - * (for example, during message expiry, skip, or clear backlog). Since the - * cursor ack set may no longer be available after mark-delete, the counter - * adjustment relies on the remaining unacked count stored in the - * {@code PendingAcksMap} entries. - * - * @param ledgerId the ledger ID of the inclusive upper bound position - * @param entryId the entry ID of the inclusive upper bound position - */ - default void prunePendingAcksUpToPosition(long ledgerId, long entryId) { - // No-op by default - } - /** * A callback hook after acknowledge messages. * @param exOfDeletion the ex of {@link org.apache.bookkeeper.mledger.ManagedCursor#asyncDelete}, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index ceef012035eda..c569cf5b68cb2 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java @@ -589,10 +589,14 @@ public CopyOnWriteArrayList getConsumers() { } @Override - public void prunePendingAcksUpToPosition(long ledgerId, long entryId) { - redeliveryMessages.removeAllUpTo(ledgerId, entryId); - for (Consumer consumer : consumerList) { - consumer.removePendingAcksUpToPositionAndDecrementUnacked(ledgerId, entryId); + public void markDeletePositionMoveForward() { + Position markDeletePosition = cursor.getMarkDeletedPosition(); + if (markDeletePosition != null) { + redeliveryMessages.removeAllUpTo(markDeletePosition.getLedgerId(), markDeletePosition.getEntryId()); + for (Consumer consumer : consumerList) { + consumer.removePendingAcksUpToPositionAndDecrementUnacked( + markDeletePosition.getLedgerId(), markDeletePosition.getEntryId()); + } } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java index bf1c6ac4dc9c4..a9f7e30510413 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java @@ -41,7 +41,6 @@ import org.apache.pulsar.broker.service.Dispatcher; 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; /** @@ -215,14 +214,7 @@ public void markDeleteComplete(Object ctx) { if (subscription != null) { Dispatcher dispatcher = subscription.getDispatcher(); if (dispatcher != null) { - Position mdPos = cursor.getMarkDeletedPosition(); - if (mdPos != null) { - dispatcher.prunePendingAcksUpToPosition(mdPos.getLedgerId(), mdPos.getEntryId()); - } - // If the subscription is a Key_Shared subscription, we should to trigger message dispatch. - if (subscription.getType() == SubType.Key_Shared) { - dispatcher.markDeletePositionMoveForward(); - } + dispatcher.markDeletePositionMoveForward(); } } expirationCheckInProgress = FALSE; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java index 2549e4a34a911..02055b28f3b53 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java @@ -642,6 +642,7 @@ protected int getStickyKeyHash(Entry entry) { @Override public void markDeletePositionMoveForward() { + 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(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java index 4a15e9a6bd42f..38fbb434691f8 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java @@ -490,6 +490,7 @@ private int getRestrictedMaxEntriesForConsumer(Consumer consumer, List { 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 8ce5ad5c897cb..eb3d024ab9a0d 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 @@ -577,10 +577,6 @@ private void notifyTheMarkDeletePositionChanged(Position oldPosition) { if (newMD.compareTo(oldPosition) != 0) { updateLastMarkDeleteAdvancedTimestamp(); handleReplicatedSubscriptionsUpdate(newMD); - - if (dispatcher != null) { - dispatcher.markDeletePositionMoveForward(); - } } } @@ -798,10 +794,7 @@ public void clearBacklogComplete(Object ctx) { future.complete(null); } }); - Position mdPos = cursor.getMarkDeletedPosition(); - if (mdPos != null) { - dispatcher.prunePendingAcksUpToPosition(mdPos.getLedgerId(), mdPos.getEntryId()); - } + dispatcher.markDeletePositionMoveForward(); dispatcher.afterAckMessages(null, ctx); } else { future.complete(null); @@ -841,10 +834,7 @@ public void skipEntriesComplete(Object ctx) { .log("Skipped messages"); future.complete(null); if (dispatcher != null) { - Position mdPos = cursor.getMarkDeletedPosition(); - if (mdPos != null) { - dispatcher.prunePendingAcksUpToPosition(mdPos.getLedgerId(), mdPos.getEntryId()); - } + dispatcher.markDeletePositionMoveForward(); dispatcher.afterAckMessages(null, ctx); } } From 83ad9bc42983e3e56c8bfd8b5ce2dde845500009 Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Wed, 29 Apr 2026 12:27:01 +0800 Subject: [PATCH 3/5] Remove comment out --- .../src/main/java/org/apache/pulsar/broker/service/Consumer.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index ff489ade39729..a198905eed90c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -610,7 +610,6 @@ private CompletableFuture individualAckNormal(CommandAck ack, Map Date: Wed, 29 Apr 2026 15:17:39 +0800 Subject: [PATCH 4/5] Fix testAckWithTransactionReduceUnackCountNotInPendingAcks --- .../pulsar/client/impl/TransactionEndToEndTest.java | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java index 90a11d34eff44..16467fabd1599 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TransactionEndToEndTest.java @@ -1527,14 +1527,7 @@ public void testAckWithTransactionReduceUnackCountNotInPendingAcks() throws Exce messageIds.add(consumer.receive(waitTimeForCanReceiveMsgInSec, TimeUnit.SECONDS).getMessageId()); } - MessageIdImpl messageId = (MessageIdImpl) messageIds.get(0); - - - // remove the message from the pendingAcks, in fact redeliver will remove the messageId from the pendingAck - getPulsarServiceList().get(0).getBrokerService().getTopic(topic, false) - .get().get().getSubscription(subName).getConsumers().get(0).getPendingAcks() - .remove(messageId.ledgerId, messageId.entryId); - + // Ack one message in the batch with a transaction. Transaction txn = getTxn(); consumer.acknowledgeAsync(messageIds.get(1), txn).get(); From 6a2ba7292b79b980a87038324103b177a2ae294e Mon Sep 17 00:00:00 2001 From: Zixuan Liu Date: Wed, 29 Apr 2026 18:08:26 +0800 Subject: [PATCH 5/5] Move markDeletePositionMoveForward call to afterAckMessages callback --- ...ersistentStickyKeyDispatcherMultipleConsumersClassic.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java index 38fbb434691f8..e5e81af4ca4f5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java @@ -489,12 +489,13 @@ private int getRestrictedMaxEntriesForConsumer(Consumer consumer, List { synchronized (PersistentStickyKeyDispatcherMultipleConsumersClassic.this) { + super.markDeletePositionMoveForward(); if (recentlyJoinedConsumers != null && !recentlyJoinedConsumers.isEmpty() && removeConsumersFromRecentJoinedConsumers()) { // After we process acks, we need to check whether the mark-delete position was advanced and we