diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java index 4167c040ed61c..7802ed07781ba 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedCursor.java @@ -167,6 +167,21 @@ void asyncReadEntries(int numberOfEntriesToRead, ReadEntriesCallback callback, O void asyncReadEntries(int numberOfEntriesToRead, long maxSizeBytes, ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition); + /** + * Asynchronously read entries from the ManagedLedger. + * + * @param numberOfEntriesToRead maximum number of entries to return + * @param maxSizeBytes max size in bytes of the entries to return + * @param callback callback object + * @param ctx opaque context + * @param maxPosition max position can read + * @param skipCondition predicate of read filter out + */ + default void asyncReadEntriesWithSkip(int numberOfEntriesToRead, long maxSizeBytes, ReadEntriesCallback callback, + Object ctx, PositionImpl maxPosition, Predicate skipCondition) { + asyncReadEntries(numberOfEntriesToRead, maxSizeBytes, callback, ctx, maxPosition); + } + /** * Get 'N'th entry from the mark delete position in the cursor without updating any cursor positions. * @@ -264,6 +279,55 @@ void asyncReadEntriesOrWait(int numberOfEntriesToRead, ReadEntriesCallback callb void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition); + /** + * Asynchronously read entries from the ManagedLedger, up to the specified number and size. + * + *

If no entries are available, the callback will not be triggered. Instead it will be registered to wait until + * a new message will be persisted into the managed ledger + * + * @see #readEntriesOrWait(int, long) + * @param maxEntries + * maximum number of entries to return + * @param callback + * callback object + * @param ctx + * opaque context + * @param maxPosition + * max position can read + * @param skipCondition + * predicate of read filter out + */ + default void asyncReadEntriesWithSkipOrWait(int maxEntries, ReadEntriesCallback callback, Object ctx, + PositionImpl maxPosition, Predicate skipCondition) { + asyncReadEntriesOrWait(maxEntries, callback, ctx, maxPosition); + } + + /** + * Asynchronously read entries from the ManagedLedger, up to the specified number and size. + * + *

If no entries are available, the callback will not be triggered. Instead it will be registered to wait until + * a new message will be persisted into the managed ledger + * + * @see #readEntriesOrWait(int, long) + * @param maxEntries + * maximum number of entries to return + * @param maxSizeBytes + * max size in bytes of the entries to return + * @param callback + * callback object + * @param ctx + * opaque context + * @param maxPosition + * max position can read + * @param skipCondition + * predicate of read filter out + */ + default void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, ReadEntriesCallback callback, + Object ctx, PositionImpl maxPosition, + Predicate skipCondition) { + asyncReadEntriesOrWait(maxEntries, maxSizeBytes, callback, ctx, maxPosition); + } + /** * Cancel a previously scheduled asyncReadEntriesOrWait operation. * diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java index 3af2e568d86f3..5b351c99649ed 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java @@ -762,13 +762,19 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { @Override public void asyncReadEntries(final int numberOfEntriesToRead, final ReadEntriesCallback callback, - final Object ctx, PositionImpl maxPosition) { + final Object ctx, PositionImpl maxPosition) { asyncReadEntries(numberOfEntriesToRead, NO_MAX_SIZE_LIMIT, callback, ctx, maxPosition); } @Override public void asyncReadEntries(int numberOfEntriesToRead, long maxSizeBytes, ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition) { + asyncReadEntriesWithSkip(numberOfEntriesToRead, maxSizeBytes, callback, ctx, maxPosition, null); + } + + @Override + public void asyncReadEntriesWithSkip(int numberOfEntriesToRead, long maxSizeBytes, ReadEntriesCallback callback, + Object ctx, PositionImpl maxPosition, Predicate skipCondition) { checkArgument(numberOfEntriesToRead > 0); if (isClosed()) { callback.readEntriesFailed(new ManagedLedgerException @@ -779,7 +785,8 @@ public void asyncReadEntries(int numberOfEntriesToRead, long maxSizeBytes, ReadE int numOfEntriesToRead = applyMaxSizeCap(numberOfEntriesToRead, maxSizeBytes); PENDING_READ_OPS_UPDATER.incrementAndGet(this); - OpReadEntry op = OpReadEntry.create(this, readPosition, numOfEntriesToRead, callback, ctx, maxPosition); + OpReadEntry op = + OpReadEntry.create(this, readPosition, numOfEntriesToRead, callback, ctx, maxPosition, skipCondition); ledger.asyncReadEntries(op); } @@ -901,6 +908,20 @@ public void asyncReadEntriesOrWait(int numberOfEntriesToRead, ReadEntriesCallbac @Override public void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition) { + asyncReadEntriesWithSkipOrWait(maxEntries, maxSizeBytes, callback, ctx, maxPosition, null); + } + + @Override + public void asyncReadEntriesWithSkipOrWait(int maxEntries, ReadEntriesCallback callback, + Object ctx, PositionImpl maxPosition, + Predicate skipCondition) { + asyncReadEntriesWithSkipOrWait(maxEntries, NO_MAX_SIZE_LIMIT, callback, ctx, maxPosition, skipCondition); + } + + @Override + public void asyncReadEntriesWithSkipOrWait(int maxEntries, long maxSizeBytes, ReadEntriesCallback callback, + Object ctx, PositionImpl maxPosition, + Predicate skipCondition) { checkArgument(maxEntries > 0); if (isClosed()) { callback.readEntriesFailed(new CursorAlreadyClosedException("Cursor was already closed"), ctx); @@ -914,10 +935,11 @@ public void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, ReadEntrie if (log.isDebugEnabled()) { log.debug("[{}] [{}] Read entries immediately", ledger.getName(), name); } - asyncReadEntries(numberOfEntriesToRead, callback, ctx, maxPosition); + asyncReadEntriesWithSkip(numberOfEntriesToRead, NO_MAX_SIZE_LIMIT, callback, ctx, + maxPosition, skipCondition); } else { OpReadEntry op = OpReadEntry.create(this, readPosition, numberOfEntriesToRead, callback, - ctx, maxPosition); + ctx, maxPosition, skipCondition); if (!WAITING_READ_OP_UPDATER.compareAndSet(this, null, op)) { op.recycle(); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java index c849a347c7d8e..ba91f605acf8c 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java @@ -2056,6 +2056,39 @@ private void internalReadFromLedger(ReadHandle ledger, OpReadEntry opReadEntry) long lastEntry = min(firstEntry + opReadEntry.getNumberOfEntriesToRead() - 1, lastEntryInLedger); + // Filer out and skip unnecessary read entry + if (opReadEntry.skipCondition != null) { + long firstValidEntry = -1L; + long lastValidEntry = -1L; + long entryId = firstEntry; + for (; entryId <= lastEntry; entryId++) { + if (opReadEntry.skipCondition.test(PositionImpl.get(ledger.getId(), entryId))) { + if (firstValidEntry == -1L) { + firstValidEntry = entryId; + } + } else { + if (firstValidEntry != -1L) { + break; + } + } + + if (firstValidEntry != -1L) { + lastValidEntry = entryId; + } + } + + // If all messages in [firstEntry...lastEntry] are filter out, + // then manual call internalReadEntriesComplete to advance read position. + if (firstValidEntry == -1L) { + opReadEntry.internalReadEntriesComplete(Collections.emptyList(), opReadEntry.ctx, + PositionImpl.get(ledger.getId(), lastEntry)); + return; + } + + firstEntry = firstValidEntry; + lastEntry = lastValidEntry; + } + if (log.isDebugEnabled()) { log.debug("[{}] Reading entries from ledger {} - first={} last={}", name, ledger.getId(), firstEntry, lastEntry); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java index 86290d8bc8f5f..04437b5320d78 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java @@ -22,7 +22,9 @@ import io.netty.util.Recycler; import io.netty.util.Recycler.Handle; import java.util.ArrayList; +import java.util.Collections; import java.util.List; +import java.util.function.Predicate; import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedLedgerException; @@ -45,8 +47,10 @@ class OpReadEntry implements ReadEntriesCallback { private PositionImpl nextReadPosition; PositionImpl maxPosition; + Predicate skipCondition; + public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPositionRef, int count, - ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition) { + ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition, Predicate skipCondition) { OpReadEntry op = RECYCLER.get(); op.readPosition = cursor.ledger.startReadOperationOnLedger(readPositionRef); op.cursor = cursor; @@ -57,13 +61,13 @@ public static OpReadEntry create(ManagedCursorImpl cursor, PositionImpl readPosi maxPosition = PositionImpl.LATEST; } op.maxPosition = maxPosition; + op.skipCondition = skipCondition; op.ctx = ctx; op.nextReadPosition = PositionImpl.get(op.readPosition); return op; } - @Override - public void readEntriesComplete(List returnedEntries, Object ctx) { + void internalReadEntriesComplete(List returnedEntries, Object ctx, PositionImpl lastPosition) { // Filter the returned entries for individual deleted messages int entriesCount = returnedEntries.size(); long entriesSize = 0; @@ -72,13 +76,19 @@ public void readEntriesComplete(List returnedEntries, Object ctx) { } cursor.updateReadStats(entriesCount, entriesSize); - final PositionImpl lastPosition = (PositionImpl) returnedEntries.get(entriesCount - 1).getPosition(); + if (entriesCount != 0) { + lastPosition = (PositionImpl) returnedEntries.get(entriesCount - 1).getPosition(); + } if (log.isDebugEnabled()) { log.debug("[{}][{}] Read entries succeeded batch_size={} cumulative_size={} requested_count={}", cursor.ledger.getName(), cursor.getName(), returnedEntries.size(), entries.size(), count); } - List filteredEntries = cursor.filterReadEntries(returnedEntries); - entries.addAll(filteredEntries); + + List filteredEntries = Collections.emptyList(); + if (entriesCount != 0) { + filteredEntries = cursor.filterReadEntries(returnedEntries); + entries.addAll(filteredEntries); + } // if entries have been filtered out then try to skip reading of already deletedMessages in that range final Position nexReadPosition = entriesCount != filteredEntries.size() @@ -87,6 +97,11 @@ public void readEntriesComplete(List returnedEntries, Object ctx) { checkReadCompletion(); } + @Override + public void readEntriesComplete(List returnedEntries, Object ctx) { + internalReadEntriesComplete(returnedEntries, ctx, null); + } + @Override public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { cursor.readOperationCompleted(); @@ -190,6 +205,7 @@ public void recycle() { nextReadPosition = null; maxPosition = null; recyclerHandle.recycle(this); + skipCondition = null; } private static final Logger log = LoggerFactory.getLogger(OpReadEntry.class); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java index 9f1cd310f92e8..6d68b042a7ad6 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java @@ -128,7 +128,7 @@ public void find() { } if (cursor.hasMoreEntries(searchPosition)) { OpReadEntry opReadEntry = OpReadEntry.create(cursor, searchPosition, batchSize, - this, OpScan.this.ctx, null); + this, OpScan.this.ctx, null, null); ledger.asyncReadEntries(opReadEntry); } else { callback.scanComplete(lastSeenPosition, ScanOutcome.COMPLETED, OpScan.this.ctx); diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java index 54ebc4c293a94..2c01b778caf6b 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorContainerTest.java @@ -33,8 +33,8 @@ import java.util.List; import java.util.Map; import java.util.Set; -import java.util.function.Predicate; import java.util.concurrent.CompletableFuture; +import java.util.function.Predicate; import org.apache.bookkeeper.mledger.AsyncCallbacks; import org.apache.bookkeeper.mledger.AsyncCallbacks.ClearBacklogCallback; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback; diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java index fa3d6ec6c9035..7e1731e573a66 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java @@ -55,6 +55,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -4126,7 +4127,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { // op readPosition is bigger than maxReadPosition OpReadEntry opReadEntry = OpReadEntry.create(cursor, ledger.lastConfirmedEntry, 10, callback, - null, PositionImpl.get(lastPosition.getLedgerId(), -1)); + null, PositionImpl.get(lastPosition.getLedgerId(), -1), null); Field field = ManagedCursorImpl.class.getDeclaredField("readPosition"); field.setAccessible(true); field.set(cursor, PositionImpl.EARLIEST); @@ -4148,7 +4149,7 @@ public void testOpReadEntryRecycle() throws Exception { }; @Cleanup final MockedStatic mockedStaticOpReadEntry = Mockito.mockStatic(OpReadEntry.class); - mockedStaticOpReadEntry.when(() -> OpReadEntry.create(any(), any(), anyInt(), any(), any(), any())) + mockedStaticOpReadEntry.when(() -> OpReadEntry.create(any(), any(), anyInt(), any(), any(), any(), any())) .thenAnswer(__ -> createOpReadEntry.get()); final ManagedLedgerConfig ledgerConfig = new ManagedLedgerConfig(); @@ -4252,5 +4253,67 @@ public void testLazyCursorLedgerCreationForSubscriptionCreation() throws Excepti factory2.shutdown(); } + @Test + public void testReadEntriesWithFilterOut() throws ManagedLedgerException, InterruptedException, ExecutionException { + int readMaxNumber = 10; + int sendNumber = 20; + ManagedLedger ledger = factory.open("testReadEntriesWithFilter"); + ManagedCursor cursor = ledger.openCursor("c"); + Position position = PositionImpl.EARLIEST; + Position maxCanReadPosition = PositionImpl.EARLIEST; + for (int i = 0; i < sendNumber; i++) { + if (i == readMaxNumber - 1) { + position = ledger.addEntry(new byte[1024]); + } else if (i == sendNumber - 1) { + maxCanReadPosition = ledger.addEntry(new byte[1024]); + } else { + ledger.addEntry(new byte[1024]); + } + + } + CompletableFuture completableFuture = new CompletableFuture<>(); + cursor.asyncReadEntriesWithSkipOrWait(sendNumber, new ReadEntriesCallback() { + @Override + public void readEntriesComplete(List entries, Object ctx) { + completableFuture.complete(entries.size()); + } + + @Override + public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { + completableFuture.completeExceptionally(exception); + } + }, null, (PositionImpl) position, pos -> { + return pos.getEntryId() % 2 != 0; + }); + + int number = completableFuture.get(); + assertEquals(number, readMaxNumber / 2); + + assertEquals(cursor.getReadPosition().getEntryId(), 10); + + CompletableFuture completableFuture2 = new CompletableFuture<>(); + cursor.asyncReadEntriesWithSkipOrWait(sendNumber, new ReadEntriesCallback() { + @Override + public void readEntriesComplete(List entries, Object ctx) { + completableFuture2.complete(entries.size()); + } + + @Override + public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { + completableFuture2.completeExceptionally(exception); + } + }, null, (PositionImpl) maxCanReadPosition, pos -> { + return pos.getEntryId() % 2 != 0; + }); + + int number2 = completableFuture2.get(); + assertEquals(number2, readMaxNumber / 2); + + assertEquals(cursor.getReadPosition().getEntryId(), 20); + + cursor.close(); + ledger.close(); + } + private static final Logger log = LoggerFactory.getLogger(ManagedCursorTest.class); } diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java index 29e30a958bd02..896d7312462e4 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java @@ -555,7 +555,7 @@ public void readEntriesComplete(List entries, Object ctx) { public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { } - }, null, maxPosition); + }, null, maxPosition, null); Assert.assertEquals(opReadEntry.readPosition, position); } @@ -3030,7 +3030,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) { responseException2.set(exception); } - }, null, PositionImpl.LATEST); + }, null, PositionImpl.LATEST, null); ledger.asyncReadEntry(ledgerHandle, PositionImpl.EARLIEST.getEntryId(), PositionImpl.EARLIEST.getEntryId(), opReadEntry, ctxStr); retryStrategically((test) -> { 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 9df30095131c3..9d07d4278fc9f 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 @@ -37,6 +37,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.function.Predicate; import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; @@ -48,6 +49,7 @@ import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.delayed.DelayedDeliveryTracker; import org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker; +import org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker; import org.apache.pulsar.broker.service.AbstractDispatcherMultipleConsumers; import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; @@ -318,8 +320,20 @@ public synchronized void readMoreEntries() { minReplayedPosition = null; } - cursor.asyncReadEntriesOrWait(messagesToRead, bytesToRead, this, - ReadType.Normal, topic.getMaxReadPosition()); + // Filter out and skip read delayed messages exist in DelayedDeliveryTracker + if (delayedDeliveryTracker.isPresent()) { + Predicate skipCondition = null; + final DelayedDeliveryTracker deliveryTracker = delayedDeliveryTracker.get(); + if (deliveryTracker instanceof BucketDelayedDeliveryTracker) { + skipCondition = position -> deliveryTracker + .containsMessage(position.getLedgerId(), position.getEntryId()); + } + cursor.asyncReadEntriesWithSkipOrWait(messagesToRead, bytesToRead, this, ReadType.Normal, + topic.getMaxReadPosition(), skipCondition); + } else { + cursor.asyncReadEntriesOrWait(messagesToRead, bytesToRead, this, ReadType.Normal, + topic.getMaxReadPosition()); + } } else { log.debug("[{}] Cannot schedule next read until previous one is done", name); }