diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index eb15e31b49b56..d78dac899b732 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -317,6 +317,9 @@ private static class EstimateTimeBasedBacklogQuotaCheckResult { Long estimatedOldestUnacknowledgedMessageTimestamp; } + // The last position that can be dispatched to consumers + private volatile Position lastDispatchablePosition; + /*** * We use 3 futures to prevent a new closing if there is an in-progress deletion or closing. We make Pulsar return * the in-progress one when it is called the second time. @@ -3792,18 +3795,57 @@ public Position getLastPosition() { @Override public CompletableFuture getLastDispatchablePosition() { - return ManagedLedgerImplUtils.asyncGetLastValidPosition((ManagedLedgerImpl) ledger, entry -> { - MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer()); - // If a messages has marker will filter by AbstractBaseDispatcher.filterEntriesForConsumer - if (Markers.isServerOnlyMarker(md)) { - return false; - } else if (md.hasTxnidMostBits() && md.hasTxnidLeastBits()) { - // Filter-out transaction aborted messages. - TxnID txnID = new TxnID(md.getTxnidMostBits(), md.getTxnidLeastBits()); - return !isTxnAborted(txnID, (PositionImpl) entry.getPosition()); - } - return true; - }, getMaxReadPosition()); + if (lastDispatchablePosition != null) { + return CompletableFuture.completedFuture(lastDispatchablePosition); + } + return ManagedLedgerImplUtils + .asyncGetLastValidPosition((ManagedLedgerImpl) ledger, entry -> { + MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer()); + // If a messages has marker will filter by AbstractBaseDispatcher.filterEntriesForConsumer + if (Markers.isServerOnlyMarker(md)) { + return false; + } else if (md.hasTxnidMostBits() && md.hasTxnidLeastBits()) { + // Filter-out transaction aborted messages. + TxnID txnID = new TxnID(md.getTxnidMostBits(), md.getTxnidLeastBits()); + return !isTxnAborted(txnID, (PositionImpl) entry.getPosition()); + } + return true; + }, getMaxReadPosition()) + .thenApply(position -> { + // Update lastDispatchablePosition to the given position + updateLastDispatchablePosition(position); + return position; + }); + } + + /** + * Update lastDispatchablePosition if the given position is greater than the lastDispatchablePosition. + * + * @param position + */ + public synchronized void updateLastDispatchablePosition(Position position) { + // Update lastDispatchablePosition to null if the position is null, fallback to + // ManagedLedgerImplUtils#asyncGetLastValidPosition + if (position == null) { + lastDispatchablePosition = null; + return; + } + + PositionImpl position0 = (PositionImpl) position; + // If the position is greater than the maxReadPosition, ignore + if (position0.compareTo(getMaxReadPosition()) > 0) { + return; + } + // If the lastDispatchablePosition is null, set it to the position + if (lastDispatchablePosition == null) { + lastDispatchablePosition = position; + return; + } + // If the position is greater than the lastDispatchablePosition, update it + PositionImpl lastDispatchablePosition0 = (PositionImpl) lastDispatchablePosition; + if (position0.compareTo(lastDispatchablePosition0) > 0) { + lastDispatchablePosition = position; + } } @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java index bab7b64c608c4..533d0716d413c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/InMemTransactionBuffer.java @@ -377,8 +377,11 @@ public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) { @Override public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) { - if (!isMarkerMessage && maxReadPositionCallBack != null) { - maxReadPositionCallBack.maxReadPositionMovedForward(null, position); + if (!isMarkerMessage) { + updateLastDispatchablePosition(position); + if (maxReadPositionCallBack != null) { + maxReadPositionCallBack.maxReadPositionMovedForward(null, position); + } } } @@ -436,4 +439,11 @@ public long getCommittedTxnCount() { .filter(txnBuffer -> txnBuffer.status.equals(TxnStatus.COMMITTED)) .count(); } + + // ThreadSafe + private void updateLastDispatchablePosition(Position position) { + if (topic instanceof PersistentTopic t) { + t.updateLastDispatchablePosition(position); + } + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java index dfb73815e08d7..fbd4ddf7da053 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java @@ -297,6 +297,11 @@ private void handleTransactionMessage(TxnID txnId, Position position) { } } + // ThreadSafe + private void updateLastDispatchablePosition(Position position) { + topic.updateLastDispatchablePosition(position); + } + @Override public CompletableFuture openTransactionBufferReader(TxnID txnID, long startSequenceId) { return null; @@ -459,6 +464,8 @@ void removeTxnAndUpdateMaxReadPosition(TxnID txnID) { } else { updateMaxReadPosition((PositionImpl) topic.getManagedLedger().getLastConfirmedEntry(), false); } + // Update the last dispatchable position to null if there is a TXN finished. + updateLastDispatchablePosition(null); } /** @@ -523,6 +530,10 @@ public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean i } } } + // If the message is a normal message, update the last dispatchable position. + if (!isMarkerMessage) { + updateLastDispatchablePosition(position); + } } @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java index ebd61dbaa82ec..6f5dc0cd4d0dd 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferDisable.java @@ -99,8 +99,11 @@ public boolean isTxnAborted(TxnID txnID, PositionImpl readPosition) { @Override public void syncMaxReadPositionForNormalPublish(PositionImpl position, boolean isMarkerMessage) { - if (!isMarkerMessage && maxReadPositionCallBack != null) { - maxReadPositionCallBack.maxReadPositionMovedForward(null, position); + if (!isMarkerMessage) { + updateLastDispatchablePosition(position); + if (maxReadPositionCallBack != null) { + maxReadPositionCallBack.maxReadPositionMovedForward(null, position); + } } } @@ -148,4 +151,11 @@ public long getAbortedTxnCount() { public long getCommittedTxnCount() { return 0; } + + // ThreadSafe + private void updateLastDispatchablePosition(Position position) { + if (topic instanceof PersistentTopic t) { + t.updateLastDispatchablePosition(position); + } + } }