diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java index 637de8a79fc1f..dc1d1eb6c9ac5 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/util/Futures.java @@ -71,14 +71,15 @@ public static CompletableFuture waitForAll(List> f } public static CompletableFuture executeWithRetry(Supplier> op, - Class needRetryExceptionClass) { + Class needRetryExceptionClass, + int maxRetryTimes) { CompletableFuture resultFuture = new CompletableFuture<>(); op.get().whenComplete((res, ex) -> { if (ex == null) { resultFuture.complete(res); } else { - if (needRetryExceptionClass.isAssignableFrom(ex.getClass())) { - executeWithRetry(op, needRetryExceptionClass).whenComplete((res2, ex2) -> { + if (needRetryExceptionClass.isAssignableFrom(ex.getClass()) && maxRetryTimes > 0) { + executeWithRetry(op, needRetryExceptionClass, maxRetryTimes - 1).whenComplete((res2, ex2) -> { if (ex2 == null) { resultFuture.complete(res2); } else { diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java index f0d4083565657..2ef8f3eae327c 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorPropertiesTest.java @@ -247,13 +247,13 @@ public void testUpdateCursorPropertiesConcurrent() throws Exception { map.put("c", "3"); futures.add(executeWithRetry(() -> c1.setCursorProperties(map), - ManagedLedgerException.BadVersionException.class)); + ManagedLedgerException.BadVersionException.class, 3)); futures.add(executeWithRetry(() -> c1.putCursorProperty("a", "2"), - ManagedLedgerException.BadVersionException.class)); + ManagedLedgerException.BadVersionException.class, 3)); futures.add(executeWithRetry(() -> c1.removeCursorProperty("c"), - ManagedLedgerException.BadVersionException.class)); + ManagedLedgerException.BadVersionException.class, 3)); for (CompletableFuture future : futures) { future.get(); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java new file mode 100644 index 0000000000000..5c99e4c307d7c --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/AbstractDelayedDeliveryTracker.java @@ -0,0 +1,162 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed; + +import io.netty.util.Timeout; +import io.netty.util.Timer; +import io.netty.util.TimerTask; +import java.time.Clock; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; + +@Slf4j +public abstract class AbstractDelayedDeliveryTracker implements DelayedDeliveryTracker, TimerTask { + + protected final PersistentDispatcherMultipleConsumers dispatcher; + + // Reference to the shared (per-broker) timer for delayed delivery + protected final Timer timer; + + // Current timeout or null if not set + protected Timeout timeout; + + // Timestamp at which the timeout is currently set + private long currentTimeoutTarget; + + // Last time the TimerTask was triggered for this class + private long lastTickRun; + + protected long tickTimeMillis; + + protected final Clock clock; + + private final boolean isDelayedDeliveryDeliverAtTimeStrict; + + public AbstractDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer, + long tickTimeMillis, + boolean isDelayedDeliveryDeliverAtTimeStrict) { + this(dispatcher, timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict); + } + + public AbstractDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer, + long tickTimeMillis, Clock clock, + boolean isDelayedDeliveryDeliverAtTimeStrict) { + this.dispatcher = dispatcher; + this.timer = timer; + this.tickTimeMillis = tickTimeMillis; + this.clock = clock; + this.isDelayedDeliveryDeliverAtTimeStrict = isDelayedDeliveryDeliverAtTimeStrict; + } + + + /** + * When {@link #isDelayedDeliveryDeliverAtTimeStrict} is false, we allow for early delivery by as much as the + * {@link #tickTimeMillis} because it is a slight optimization to let messages skip going back into the delay + * tracker for a brief amount of time when we're already trying to dispatch to the consumer. + * + * When {@link #isDelayedDeliveryDeliverAtTimeStrict} is true, we use the current time to determine when messages + * can be delivered. As a consequence, there are two delays that will affect delivery. The first is the + * {@link #tickTimeMillis} and the second is the {@link Timer}'s granularity. + * + * @return the cutoff time to determine whether a message is ready to deliver to the consumer + */ + protected long getCutoffTime() { + return isDelayedDeliveryDeliverAtTimeStrict ? clock.millis() : clock.millis() + tickTimeMillis; + } + + public void resetTickTime(long tickTime) { + if (this.tickTimeMillis != tickTime) { + this.tickTimeMillis = tickTime; + } + } + + protected void updateTimer() { + if (getNumberOfDelayedMessages() == 0) { + if (timeout != null) { + currentTimeoutTarget = -1; + timeout.cancel(); + timeout = null; + } + return; + } + long timestamp = nextDeliveryTime(); + if (timestamp == currentTimeoutTarget) { + // The timer is already set to the correct target time + return; + } + + if (timeout != null) { + timeout.cancel(); + } + + long now = clock.millis(); + long delayMillis = timestamp - now; + + if (delayMillis < 0) { + // There are messages that are already ready to be delivered. If + // the dispatcher is not getting them is because the consumer is + // either not connected or slow. + // We don't need to keep retriggering the timer. When the consumer + // catches up, the dispatcher will do the readMoreEntries() and + // get these messages + return; + } + + // Compute the earliest time that we schedule the timer to run. + long remainingTickDelayMillis = lastTickRun + tickTimeMillis - now; + long calculatedDelayMillis = Math.max(delayMillis, remainingTickDelayMillis); + + if (log.isDebugEnabled()) { + log.debug("[{}] Start timer in {} millis", dispatcher.getName(), calculatedDelayMillis); + } + + // Even though we may delay longer than this timestamp because of the tick delay, we still track the + // current timeout with reference to the next message's timestamp. + currentTimeoutTarget = timestamp; + timeout = timer.newTimeout(this, calculatedDelayMillis, TimeUnit.MILLISECONDS); + } + + @Override + public void run(Timeout timeout) throws Exception { + if (log.isDebugEnabled()) { + log.debug("[{}] Timer triggered", dispatcher.getName()); + } + if (timeout == null || timeout.isCancelled()) { + return; + } + + synchronized (dispatcher) { + lastTickRun = clock.millis(); + currentTimeoutTarget = -1; + this.timeout = null; + dispatcher.readMoreEntries(); + } + } + + @Override + public void close() { + if (timeout != null) { + timeout.cancel(); + timeout = null; + } + } + + protected abstract long nextDeliveryTime(); +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java index da28ff19234b4..f55d5fd11694b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java @@ -18,47 +18,28 @@ */ package org.apache.pulsar.broker.delayed; -import io.netty.util.Timeout; +import com.google.common.annotations.VisibleForTesting; import io.netty.util.Timer; -import io.netty.util.TimerTask; import java.time.Clock; import java.util.NavigableSet; import java.util.TreeSet; -import java.util.concurrent.TimeUnit; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue; @Slf4j -public class InMemoryDelayedDeliveryTracker implements DelayedDeliveryTracker, TimerTask { +public class InMemoryDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker { protected final TripleLongPriorityQueue priorityQueue = new TripleLongPriorityQueue(); - private final PersistentDispatcherMultipleConsumers dispatcher; - - // Reference to the shared (per-broker) timer for delayed delivery - private final Timer timer; - - // Current timeout or null if not set - protected Timeout timeout; - - // Timestamp at which the timeout is currently set - private long currentTimeoutTarget; - - // Last time the TimerTask was triggered for this class - private long lastTickRun; - - private long tickTimeMillis; - - private final Clock clock; - - private final boolean isDelayedDeliveryDeliverAtTimeStrict; - // If we detect that all messages have fixed delay time, such that the delivery is // always going to be in FIFO order, then we can avoid pulling all the messages in // tracker. Instead, we use the lookahead for detection and pause the read from // the cursor if the delays are fixed. + @Getter + @VisibleForTesting private final long fixedDelayDetectionLookahead; // This is the timestamp of the message with the highest delivery time @@ -76,33 +57,14 @@ public class InMemoryDelayedDeliveryTracker implements DelayedDeliveryTracker, T fixedDelayDetectionLookahead); } - InMemoryDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer, + public InMemoryDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer, long tickTimeMillis, Clock clock, boolean isDelayedDeliveryDeliverAtTimeStrict, long fixedDelayDetectionLookahead) { - this.dispatcher = dispatcher; - this.timer = timer; - this.tickTimeMillis = tickTimeMillis; - this.clock = clock; - this.isDelayedDeliveryDeliverAtTimeStrict = isDelayedDeliveryDeliverAtTimeStrict; + super(dispatcher, timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict); this.fixedDelayDetectionLookahead = fixedDelayDetectionLookahead; } - /** - * When {@link #isDelayedDeliveryDeliverAtTimeStrict} is false, we allow for early delivery by as much as the - * {@link #tickTimeMillis} because it is a slight optimization to let messages skip going back into the delay - * tracker for a brief amount of time when we're already trying to dispatch to the consumer. - * - * When {@link #isDelayedDeliveryDeliverAtTimeStrict} is true, we use the current time to determine when messages - * can be delivered. As a consequence, there are two delays that will affect delivery. The first is the - * {@link #tickTimeMillis} and the second is the {@link Timer}'s granularity. - * - * @return the cutoff time to determine whether a message is ready to deliver to the consumer - */ - private long getCutoffTime() { - return isDelayedDeliveryDeliverAtTimeStrict ? clock.millis() : clock.millis() + tickTimeMillis; - } - @Override public boolean addMessage(long ledgerId, long entryId, long deliverAt) { if (deliverAt < 0 || deliverAt <= getCutoffTime()) { @@ -115,19 +77,24 @@ public boolean addMessage(long ledgerId, long entryId, long deliverAt) { deliverAt - clock.millis()); } - priorityQueue.add(deliverAt, ledgerId, entryId); updateTimer(); - // Check that new delivery time comes after the current highest, or at - // least within a single tick time interval of 1 second. + checkAndUpdateHighest(deliverAt); + + return true; + } + + /** + * Check that new delivery time comes after the current highest, or at + * least within a single tick time interval of 1 second. + */ + private void checkAndUpdateHighest(long deliverAt) { if (deliverAt < (highestDeliveryTimeTracked - tickTimeMillis)) { messagesHaveFixedDelay = false; } highestDeliveryTimeTracked = Math.max(highestDeliveryTimeTracked, deliverAt); - - return true; } /** @@ -179,14 +146,6 @@ public NavigableSet getScheduledMessages(int maxMessages) { return positions; } - @Override - public void resetTickTime(long tickTime) { - - if (this.tickTimeMillis != tickTime) { - this.tickTimeMillis = tickTime; - } - } - @Override public void clear() { this.priorityQueue.clear(); @@ -202,87 +161,9 @@ public long getBufferMemoryUsage() { return priorityQueue.bytesCapacity(); } - /** - * Update the scheduled timer task such that: - * 1. If there are no delayed messages, return and do not schedule a timer task. - * 2. If the next message in the queue has the same deliverAt time as the timer task, return and leave existing - * timer task in place. - * 3. If the deliverAt time for the next delayed message has already passed (i.e. the delay is negative), return - * without scheduling a timer task since the subscription is backlogged. - * 4. Else, schedule a timer task where the delay is the greater of these two: the next message's deliverAt time or - * the last tick time plus the tickTimeMillis (to ensure we do not schedule the task more frequently than the - * tickTimeMillis). - */ - private void updateTimer() { - if (priorityQueue.isEmpty()) { - if (timeout != null) { - currentTimeoutTarget = -1; - timeout.cancel(); - timeout = null; - } - return; - } - - long timestamp = priorityQueue.peekN1(); - if (timestamp == currentTimeoutTarget) { - // The timer is already set to the correct target time - return; - } - - if (timeout != null) { - timeout.cancel(); - } - - long now = clock.millis(); - long delayMillis = timestamp - now; - - if (delayMillis < 0) { - // There are messages that are already ready to be delivered. If - // the dispatcher is not getting them is because the consumer is - // either not connected or slow. - // We don't need to keep retriggering the timer. When the consumer - // catches up, the dispatcher will do the readMoreEntries() and - // get these messages - return; - } - - // Compute the earliest time that we schedule the timer to run. - long remainingTickDelayMillis = lastTickRun + tickTimeMillis - now; - long calculatedDelayMillis = Math.max(delayMillis, remainingTickDelayMillis); - - if (log.isDebugEnabled()) { - log.debug("[{}] Start timer in {} millis", dispatcher.getName(), calculatedDelayMillis); - } - - // Even though we may delay longer than this timestamp because of the tick delay, we still track the - // current timeout with reference to the next message's timestamp. - currentTimeoutTarget = timestamp; - timeout = timer.newTimeout(this, calculatedDelayMillis, TimeUnit.MILLISECONDS); - } - - @Override - public void run(Timeout timeout) throws Exception { - if (log.isDebugEnabled()) { - log.debug("[{}] Timer triggered", dispatcher.getName()); - } - if (timeout == null || timeout.isCancelled()) { - return; - } - - synchronized (dispatcher) { - lastTickRun = clock.millis(); - currentTimeoutTarget = -1; - this.timeout = null; - dispatcher.readMoreEntries(); - } - } - @Override public void close() { - if (timeout != null) { - timeout.cancel(); - timeout = null; - } + super.close(); priorityQueue.close(); } @@ -291,7 +172,7 @@ public boolean shouldPauseAllDeliveries() { // Pause deliveries if we know all delays are fixed within the lookahead window return fixedDelayDetectionLookahead > 0 && messagesHaveFixedDelay - && priorityQueue.size() >= fixedDelayDetectionLookahead + && getNumberOfDelayedMessages() >= fixedDelayDetectionLookahead && !hasMessageAvailable(); } @@ -299,4 +180,8 @@ public boolean shouldPauseAllDeliveries() { public boolean containsMessage(long ledgerId, long entryId) { return false; } + + protected long nextDeliveryTime() { + return priorityQueue.peekN1(); + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java new file mode 100644 index 0000000000000..fbd6d765705d4 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/Bucket.java @@ -0,0 +1,145 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed.bucket; + +import static org.apache.bookkeeper.mledger.util.Futures.executeWithRetry; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat; +import org.roaringbitmap.RoaringBitmap; + +@Slf4j +@Data +@AllArgsConstructor +abstract class Bucket { + + static final String DELAYED_BUCKET_KEY_PREFIX = "#pulsar.internal.delayed.bucket"; + static final String DELIMITER = "_"; + static final int MaxRetryTimes = 3; + + protected final ManagedCursor cursor; + protected final BucketSnapshotStorage bucketSnapshotStorage; + + long startLedgerId; + long endLedgerId; + + Map delayedIndexBitMap; + + long numberBucketDelayedMessages; + + int lastSegmentEntryId; + + int currentSegmentEntryId; + + long snapshotLength; + + private volatile Long bucketId; + + private volatile CompletableFuture snapshotCreateFuture; + + + Bucket(ManagedCursor cursor, BucketSnapshotStorage storage, long startLedgerId, long endLedgerId) { + this(cursor, storage, startLedgerId, endLedgerId, new HashMap<>(), -1, -1, 0, 0, null, null); + } + + boolean containsMessage(long ledgerId, long entryId) { + RoaringBitmap bitSet = delayedIndexBitMap.get(ledgerId); + if (bitSet == null) { + return false; + } + return bitSet.contains(entryId, entryId + 1); + } + + void putIndexBit(long ledgerId, long entryId) { + delayedIndexBitMap.computeIfAbsent(ledgerId, k -> new RoaringBitmap()).add(entryId, entryId + 1); + } + + boolean removeIndexBit(long ledgerId, long entryId) { + boolean contained = false; + RoaringBitmap bitSet = delayedIndexBitMap.get(ledgerId); + if (bitSet != null && bitSet.contains(entryId, entryId + 1)) { + contained = true; + bitSet.remove(entryId, entryId + 1); + + if (bitSet.isEmpty()) { + delayedIndexBitMap.remove(ledgerId); + } + + if (numberBucketDelayedMessages > 0) { + numberBucketDelayedMessages--; + } + } + return contained; + } + + String bucketKey() { + return String.join(DELIMITER, DELAYED_BUCKET_KEY_PREFIX, String.valueOf(startLedgerId), + String.valueOf(endLedgerId)); + } + + Optional> getSnapshotCreateFuture() { + return Optional.ofNullable(snapshotCreateFuture); + } + + Optional getBucketId() { + return Optional.ofNullable(bucketId); + } + + long getAndUpdateBucketId() { + Optional bucketIdOptional = getBucketId(); + if (bucketIdOptional.isPresent()) { + return bucketIdOptional.get(); + } + + String bucketIdStr = cursor.getCursorProperties().get(bucketKey()); + long bucketId = Long.parseLong(bucketIdStr); + setBucketId(bucketId); + return bucketId; + } + + CompletableFuture asyncSaveBucketSnapshot( + ImmutableBucket bucketState, DelayedMessageIndexBucketSnapshotFormat.SnapshotMetadata snapshotMetadata, + List bucketSnapshotSegments) { + + return bucketSnapshotStorage.createBucketSnapshot(snapshotMetadata, bucketSnapshotSegments) + .thenCompose(newBucketId -> { + bucketState.setBucketId(newBucketId); + String bucketKey = bucketState.bucketKey(); + return putBucketKeyId(bucketKey, newBucketId).exceptionally(ex -> { + log.warn("Failed to record bucketId to cursor property, bucketKey: {}", bucketKey); + return null; + }).thenApply(__ -> newBucketId); + }); + } + + private CompletableFuture putBucketKeyId(String bucketKey, Long bucketId) { + Objects.requireNonNull(bucketId); + return executeWithRetry(() -> cursor.putCursorProperty(bucketKey, String.valueOf(bucketId)), + ManagedLedgerException.BadVersionException.class, MaxRetryTimes); + } +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java new file mode 100644 index 0000000000000..b7f0e0a1bc1e5 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed.bucket; + +import static com.google.common.base.Preconditions.checkArgument; +import com.google.common.collect.HashBasedTable; +import com.google.common.collect.Range; +import com.google.common.collect.RangeMap; +import com.google.common.collect.Table; +import com.google.common.collect.TreeRangeMap; +import io.netty.util.Timeout; +import io.netty.util.Timer; +import java.time.Clock; +import java.util.Iterator; +import java.util.NavigableSet; +import java.util.Optional; +import java.util.TreeSet; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import javax.annotation.concurrent.ThreadSafe; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.pulsar.broker.delayed.AbstractDelayedDeliveryTracker; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.DelayedIndex; +import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; +import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue; + +@Slf4j +@ThreadSafe +public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker { + + static final int AsyncOperationTimeoutSeconds = 30; + + private final long minIndexCountPerBucket; + + private final long timeStepPerBucketSnapshotSegment; + + private final int maxNumBuckets; + + private long numberDelayedMessages; + + private final MutableBucket lastMutableBucket; + + private final TripleLongPriorityQueue sharedBucketPriorityQueue; + + private final RangeMap immutableBuckets; + + private final Table snapshotSegmentLastIndexTable; + + public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, + Timer timer, long tickTimeMillis, + boolean isDelayedDeliveryDeliverAtTimeStrict, + BucketSnapshotStorage bucketSnapshotStorage, + long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegment, + int maxNumBuckets) { + this(dispatcher, timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict, + bucketSnapshotStorage, minIndexCountPerBucket, timeStepPerBucketSnapshotSegment, maxNumBuckets); + } + + public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, + Timer timer, long tickTimeMillis, Clock clock, + boolean isDelayedDeliveryDeliverAtTimeStrict, + BucketSnapshotStorage bucketSnapshotStorage, + long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegment, + int maxNumBuckets) { + super(dispatcher, timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict); + this.minIndexCountPerBucket = minIndexCountPerBucket; + this.timeStepPerBucketSnapshotSegment = timeStepPerBucketSnapshotSegment; + this.maxNumBuckets = maxNumBuckets; + this.sharedBucketPriorityQueue = new TripleLongPriorityQueue(); + this.immutableBuckets = TreeRangeMap.create(); + this.snapshotSegmentLastIndexTable = HashBasedTable.create(); + this.numberDelayedMessages = 0L; + ManagedCursor cursor = dispatcher.getCursor(); + this.lastMutableBucket = new MutableBucket(cursor, bucketSnapshotStorage); + } + + @Override + public void run(Timeout timeout) throws Exception { + synchronized (this) { + if (timeout == null || timeout.isCancelled()) { + return; + } + lastMutableBucket.moveScheduledMessageToSharedQueue(getCutoffTime(), sharedBucketPriorityQueue); + } + super.run(timeout); + } + + private Optional findImmutableBucket(long ledgerId) { + if (immutableBuckets.asMapOfRanges().isEmpty()) { + return Optional.empty(); + } + + return Optional.ofNullable(immutableBuckets.get(ledgerId)); + } + + private void sealBucket() { + Pair immutableBucketDelayedIndexPair = + lastMutableBucket.sealBucketAndAsyncPersistent(this.timeStepPerBucketSnapshotSegment, + this.sharedBucketPriorityQueue); + if (immutableBucketDelayedIndexPair != null) { + ImmutableBucket immutableBucket = immutableBucketDelayedIndexPair.getLeft(); + immutableBuckets.put(Range.closed(immutableBucket.startLedgerId, immutableBucket.endLedgerId), + immutableBucket); + + DelayedIndex lastDelayedIndex = immutableBucketDelayedIndexPair.getRight(); + snapshotSegmentLastIndexTable.put(lastDelayedIndex.getLedgerId(), lastDelayedIndex.getEntryId(), + immutableBucket); + if (log.isDebugEnabled()) { + log.debug("[{}] Create bucket snapshot, bucket: {}", dispatcher.getName(), + lastMutableBucket); + } + } + } + + @Override + public synchronized boolean addMessage(long ledgerId, long entryId, long deliverAt) { + if (containsMessage(ledgerId, entryId)) { + return true; + } + + if (deliverAt < 0 || deliverAt <= getCutoffTime()) { + return false; + } + + boolean existBucket = findImmutableBucket(ledgerId).isPresent(); + + // Create bucket snapshot + if (!existBucket && ledgerId > lastMutableBucket.endLedgerId + && lastMutableBucket.size() >= minIndexCountPerBucket + && !lastMutableBucket.isEmpty()) { + sealBucket(); + lastMutableBucket.resetLastMutableBucketRange(); + + if (immutableBuckets.asMapOfRanges().size() > maxNumBuckets) { + // TODO merge bucket snapshot (synchronize operate) + } + } + + if (ledgerId < lastMutableBucket.startLedgerId || existBucket) { + // If (ledgerId < startLedgerId || existBucket) means that message index belong to previous bucket range, + // enter sharedBucketPriorityQueue directly + sharedBucketPriorityQueue.add(deliverAt, ledgerId, entryId); + } else { + checkArgument(ledgerId >= lastMutableBucket.endLedgerId); + lastMutableBucket.addMessage(ledgerId, entryId, deliverAt); + } + + numberDelayedMessages++; + + if (log.isDebugEnabled()) { + log.debug("[{}] Add message {}:{} -- Delivery in {} ms ", dispatcher.getName(), ledgerId, entryId, + deliverAt - clock.millis()); + } + + updateTimer(); + + return true; + } + + @Override + public synchronized boolean hasMessageAvailable() { + long cutoffTime = getCutoffTime(); + + boolean hasMessageAvailable = getNumberOfDelayedMessages() > 0 && nextDeliveryTime() <= cutoffTime; + if (!hasMessageAvailable) { + updateTimer(); + } + return hasMessageAvailable; + } + + @Override + protected long nextDeliveryTime() { + if (lastMutableBucket.isEmpty() && !sharedBucketPriorityQueue.isEmpty()) { + return sharedBucketPriorityQueue.peekN1(); + } else if (sharedBucketPriorityQueue.isEmpty() && !lastMutableBucket.isEmpty()) { + return lastMutableBucket.nextDeliveryTime(); + } + long timestamp = lastMutableBucket.nextDeliveryTime(); + long bucketTimestamp = sharedBucketPriorityQueue.peekN1(); + return Math.min(timestamp, bucketTimestamp); + } + + @Override + public synchronized long getNumberOfDelayedMessages() { + return numberDelayedMessages; + } + + @Override + public synchronized long getBufferMemoryUsage() { + return this.lastMutableBucket.getBufferMemoryUsage() + sharedBucketPriorityQueue.bytesCapacity(); + } + + @Override + public synchronized NavigableSet getScheduledMessages(int maxMessages) { + long cutoffTime = getCutoffTime(); + + lastMutableBucket.moveScheduledMessageToSharedQueue(cutoffTime, sharedBucketPriorityQueue); + + NavigableSet positions = new TreeSet<>(); + int n = maxMessages; + + while (n > 0 && !sharedBucketPriorityQueue.isEmpty()) { + long timestamp = sharedBucketPriorityQueue.peekN1(); + if (timestamp > cutoffTime) { + break; + } + + long ledgerId = sharedBucketPriorityQueue.peekN2(); + long entryId = sharedBucketPriorityQueue.peekN3(); + positions.add(new PositionImpl(ledgerId, entryId)); + + sharedBucketPriorityQueue.pop(); + removeIndexBit(ledgerId, entryId); + + ImmutableBucket bucket = snapshotSegmentLastIndexTable.remove(ledgerId, entryId); + if (bucket != null) { + if (log.isDebugEnabled()) { + log.debug("[{}] Load next snapshot segment, bucket: {}", dispatcher.getName(), bucket); + } + // All message of current snapshot segment are scheduled, load next snapshot segment + // TODO make it asynchronous and not blocking this process + try { + bucket.asyncLoadNextBucketSnapshotEntry(false).thenAccept(indexList -> { + if (CollectionUtils.isEmpty(indexList)) { + return; + } + DelayedMessageIndexBucketSnapshotFormat.DelayedIndex + lastDelayedIndex = indexList.get(indexList.size() - 1); + this.snapshotSegmentLastIndexTable.put(lastDelayedIndex.getLedgerId(), + lastDelayedIndex.getEntryId(), bucket); + for (DelayedMessageIndexBucketSnapshotFormat.DelayedIndex index : indexList) { + sharedBucketPriorityQueue.add(index.getTimestamp(), index.getLedgerId(), + index.getEntryId()); + } + }).get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS); + } catch (InterruptedException | ExecutionException | TimeoutException e) { + // TODO make this segment load again + throw new RuntimeException(e); + } + } + + --n; + --numberDelayedMessages; + } + + updateTimer(); + + return positions; + } + + @Override + public boolean shouldPauseAllDeliveries() { + return false; + } + + @Override + public synchronized void clear() { + cleanImmutableBuckets(true); + sharedBucketPriorityQueue.clear(); + lastMutableBucket.clear(); + snapshotSegmentLastIndexTable.clear(); + numberDelayedMessages = 0; + } + + @Override + public synchronized void close() { + super.close(); + lastMutableBucket.close(); + cleanImmutableBuckets(false); + sharedBucketPriorityQueue.close(); + } + + private void cleanImmutableBuckets(boolean delete) { + if (immutableBuckets != null) { + Iterator iterator = immutableBuckets.asMapOfRanges().values().iterator(); + while (iterator.hasNext()) { + ImmutableBucket bucket = iterator.next(); + bucket.clear(delete); + iterator.remove(); + } + } + } + + private boolean removeIndexBit(long ledgerId, long entryId) { + if (lastMutableBucket.removeIndexBit(ledgerId, entryId)) { + return true; + } + + return findImmutableBucket(ledgerId).map(bucket -> bucket.removeIndexBit(ledgerId, entryId)) + .orElse(false); + } + + @Override + public boolean containsMessage(long ledgerId, long entryId) { + if (lastMutableBucket.containsMessage(ledgerId, entryId)) { + return true; + } + + return findImmutableBucket(ledgerId).map(bucket -> bucket.containsMessage(ledgerId, entryId)) + .orElse(false); + } +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketSnapshotStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketSnapshotStorage.java similarity index 98% rename from pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketSnapshotStorage.java rename to pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketSnapshotStorage.java index 7e5fa633dd914..3ab4ce1ad2792 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketSnapshotStorage.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketSnapshotStorage.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.pulsar.broker.delayed; +package org.apache.pulsar.broker.delayed.bucket; import java.util.List; import java.util.concurrent.CompletableFuture; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java new file mode 100644 index 0000000000000..833030c575172 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed.bucket; + +import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.AsyncOperationTimeoutSeconds; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.commons.collections4.CollectionUtils; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.DelayedIndex; + +@Slf4j +class ImmutableBucket extends Bucket { + ImmutableBucket(ManagedCursor cursor, BucketSnapshotStorage storage, long startLedgerId, long endLedgerId) { + super(cursor, storage, startLedgerId, endLedgerId); + } + + /** + * Asynchronous load next bucket snapshot entry. + * @param isRecover whether used to recover bucket snapshot + * @return CompletableFuture + */ + CompletableFuture> asyncLoadNextBucketSnapshotEntry(boolean isRecover) { + if (log.isDebugEnabled()) { + log.debug("[{}] Load next bucket snapshot data, bucket: {}", cursor.getName(), this); + } + + // Wait bucket snapshot create finish + CompletableFuture snapshotCreateFuture = + getSnapshotCreateFuture().orElseGet(() -> CompletableFuture.completedFuture(null)) + .thenApply(__ -> null); + + return snapshotCreateFuture.thenCompose(__ -> { + final long bucketId = getAndUpdateBucketId(); + CompletableFuture loadMetaDataFuture = new CompletableFuture<>(); + if (isRecover) { + // TODO Recover bucket snapshot + } else { + loadMetaDataFuture.complete(currentSegmentEntryId + 1); + } + + return loadMetaDataFuture.thenCompose(nextSegmentEntryId -> { + if (nextSegmentEntryId > lastSegmentEntryId) { + // TODO Delete bucket snapshot + return CompletableFuture.completedFuture(null); + } + + return bucketSnapshotStorage.getBucketSnapshotSegment(bucketId, nextSegmentEntryId, nextSegmentEntryId) + .thenApply(bucketSnapshotSegments -> { + if (CollectionUtils.isEmpty(bucketSnapshotSegments)) { + return Collections.emptyList(); + } + + DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment snapshotSegment = + bucketSnapshotSegments.get(0); + List indexList = + snapshotSegment.getIndexesList(); + this.setCurrentSegmentEntryId(nextSegmentEntryId); + return indexList; + }); + }); + }); + } + + void clear(boolean delete) { + delayedIndexBitMap.clear(); + getSnapshotCreateFuture().ifPresent(snapshotGenerateFuture -> { + if (delete) { + snapshotGenerateFuture.cancel(true); + // TODO delete bucket snapshot + } else { + try { + snapshotGenerateFuture.get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS); + } catch (Exception e) { + log.warn("Failed wait to snapshot generate, bucketId: {}, bucketKey: {}", getBucketId(), + bucketKey()); + } + } + }); + } +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java new file mode 100644 index 0000000000000..36026298269d7 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java @@ -0,0 +1,203 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed.bucket; + +import static com.google.common.base.Preconditions.checkArgument; +import com.google.protobuf.ByteString; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.DelayedIndex; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.SnapshotMetadata; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata; +import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue; +import org.roaringbitmap.RoaringBitmap; + +@Slf4j +class MutableBucket extends Bucket implements AutoCloseable { + + private final TripleLongPriorityQueue priorityQueue; + + MutableBucket(ManagedCursor cursor, + BucketSnapshotStorage bucketSnapshotStorage) { + super(cursor, bucketSnapshotStorage, -1L, -1L); + this.priorityQueue = new TripleLongPriorityQueue(); + } + + Pair sealBucketAndAsyncPersistent( + long timeStepPerBucketSnapshotSegment, + TripleLongPriorityQueue sharedQueue) { + if (priorityQueue.isEmpty()) { + return null; + } + long numMessages = 0; + + final long startLedgerId = getStartLedgerId(); + final long endLedgerId = getEndLedgerId(); + + List bucketSnapshotSegments = new ArrayList<>(); + List segmentMetadataList = new ArrayList<>(); + Map bitMap = new HashMap<>(); + SnapshotSegment.Builder snapshotSegmentBuilder = SnapshotSegment.newBuilder(); + SnapshotSegmentMetadata.Builder segmentMetadataBuilder = SnapshotSegmentMetadata.newBuilder(); + + long currentTimestampUpperLimit = 0; + while (!priorityQueue.isEmpty()) { + long timestamp = priorityQueue.peekN1(); + if (currentTimestampUpperLimit == 0) { + currentTimestampUpperLimit = timestamp + timeStepPerBucketSnapshotSegment - 1; + } + + long ledgerId = priorityQueue.peekN2(); + long entryId = priorityQueue.peekN3(); + + checkArgument(ledgerId >= startLedgerId && ledgerId <= endLedgerId); + + // Move first segment of bucket snapshot to sharedBucketPriorityQueue + if (segmentMetadataList.size() == 0) { + sharedQueue.add(timestamp, ledgerId, entryId); + } + + priorityQueue.pop(); + numMessages++; + + DelayedIndex delayedIndex = DelayedIndex.newBuilder() + .setTimestamp(timestamp) + .setLedgerId(ledgerId) + .setEntryId(entryId).build(); + + bitMap.computeIfAbsent(ledgerId, k -> new RoaringBitmap()).add(entryId, entryId + 1); + + snapshotSegmentBuilder.addIndexes(delayedIndex); + + if (priorityQueue.isEmpty() || priorityQueue.peekN1() > currentTimestampUpperLimit) { + segmentMetadataBuilder.setMaxScheduleTimestamp(timestamp); + currentTimestampUpperLimit = 0; + + Iterator> iterator = bitMap.entrySet().iterator(); + while (iterator.hasNext()) { + Map.Entry entry = iterator.next(); + byte[] array = new byte[entry.getValue().serializedSizeInBytes()]; + entry.getValue().serialize(ByteBuffer.wrap(array)); + segmentMetadataBuilder.putDelayedIndexBitMap(entry.getKey(), ByteString.copyFrom(array)); + iterator.remove(); + } + + segmentMetadataList.add(segmentMetadataBuilder.build()); + segmentMetadataBuilder.clear(); + + bucketSnapshotSegments.add(snapshotSegmentBuilder.build()); + snapshotSegmentBuilder.clear(); + } + } + + SnapshotMetadata bucketSnapshotMetadata = SnapshotMetadata.newBuilder() + .addAllMetadataList(segmentMetadataList) + .build(); + + final int lastSegmentEntryId = segmentMetadataList.size(); + + ImmutableBucket bucket = new ImmutableBucket(cursor, bucketSnapshotStorage, startLedgerId, endLedgerId); + bucket.setCurrentSegmentEntryId(1); + bucket.setNumberBucketDelayedMessages(numMessages); + bucket.setLastSegmentEntryId(lastSegmentEntryId); + + // Add the first snapshot segment last message to snapshotSegmentLastMessageTable + checkArgument(!bucketSnapshotSegments.isEmpty()); + SnapshotSegment snapshotSegment = bucketSnapshotSegments.get(0); + DelayedIndex lastDelayedIndex = snapshotSegment.getIndexes(snapshotSegment.getIndexesCount() - 1); + Pair result = Pair.of(bucket, lastDelayedIndex); + + CompletableFuture future = asyncSaveBucketSnapshot(bucket, + bucketSnapshotMetadata, bucketSnapshotSegments); + bucket.setSnapshotCreateFuture(future); + future.whenComplete((__, ex) -> { + if (ex == null) { + bucket.setSnapshotCreateFuture(null); + } else { + //TODO Record create snapshot failed + log.error("Failed to create snapshot: ", ex); + } + }); + + return result; + } + + void moveScheduledMessageToSharedQueue(long cutoffTime, TripleLongPriorityQueue sharedBucketPriorityQueue) { + while (!priorityQueue.isEmpty()) { + long timestamp = priorityQueue.peekN1(); + if (timestamp > cutoffTime) { + break; + } + + long ledgerId = priorityQueue.peekN2(); + long entryId = priorityQueue.peekN3(); + sharedBucketPriorityQueue.add(timestamp, ledgerId, entryId); + + priorityQueue.pop(); + } + } + + void resetLastMutableBucketRange() { + this.startLedgerId = -1L; + this.endLedgerId = -1L; + } + + void clear() { + this.resetLastMutableBucketRange(); + this.delayedIndexBitMap.clear(); + } + + public void close() { + priorityQueue.close(); + } + + long getBufferMemoryUsage() { + return priorityQueue.bytesCapacity(); + } + + boolean isEmpty() { + return priorityQueue.isEmpty(); + } + + long nextDeliveryTime() { + return priorityQueue.peekN1(); + } + + long size() { + return priorityQueue.size(); + } + + void addMessage(long ledgerId, long entryId, long deliverAt) { + priorityQueue.add(deliverAt, ledgerId, entryId); + if (startLedgerId == -1L) { + this.startLedgerId = ledgerId; + } + this.endLedgerId = ledgerId; + putIndexBit(ledgerId, entryId); + } +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/package-info.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/package-info.java new file mode 100644 index 0000000000000..0ab45ecded2cf --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/package-info.java @@ -0,0 +1,19 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed.bucket; 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 b777087ba2f39..193ee07561d2d 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 @@ -1120,6 +1120,10 @@ public long getDelayedTrackerMemoryUsage() { return 0; } + public ManagedCursor getCursor() { + return cursor; + } + protected int getStickyKeyHash(Entry entry) { return StickyKeyConsumerSelector.makeStickyKeyHash(peekStickyKey(entry.getDataBuffer())); } diff --git a/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto b/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto index eda9a8f92e670..8414a583fe5b0 100644 --- a/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto +++ b/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto @@ -24,8 +24,8 @@ option optimize_for = SPEED; message DelayedIndex { required uint64 timestamp = 1; - required int64 ledger_id = 2; - required int64 entry_id = 3; + required uint64 ledger_id = 2; + required uint64 entry_id = 3; } message SnapshotSegmentMetadata { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/AbstractDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/AbstractDeliveryTrackerTest.java new file mode 100644 index 0000000000000..1d166a8db5c9e --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/AbstractDeliveryTrackerTest.java @@ -0,0 +1,240 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed; + +import static org.mockito.Mockito.atMostOnce; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timeout; +import io.netty.util.Timer; +import io.netty.util.TimerTask; +import io.netty.util.concurrent.DefaultThreadFactory; +import java.time.Clock; +import java.util.Collections; +import java.util.NavigableMap; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; +import org.awaitility.Awaitility; +import org.testng.annotations.AfterClass; +import org.testng.annotations.Test; + +public abstract class AbstractDeliveryTrackerTest { + + // Create a single shared timer for the test. + protected final Timer timer = + new HashedWheelTimer(new DefaultThreadFactory("pulsar-in-memory-delayed-delivery-test"), + 500, TimeUnit.MILLISECONDS); + protected PersistentDispatcherMultipleConsumers dispatcher; + protected Clock clock; + + protected AtomicLong clockTime; + + @AfterClass(alwaysRun = true) + public void cleanup() { + timer.stop(); + } + + @Test(dataProvider = "delayedTracker") + public void test(DelayedDeliveryTracker tracker) throws Exception { + assertFalse(tracker.hasMessageAvailable()); + + assertTrue(tracker.addMessage(1, 2, 20)); + assertTrue(tracker.addMessage(2, 1, 10)); + assertTrue(tracker.addMessage(3, 3, 30)); + assertTrue(tracker.addMessage(4, 5, 50)); + assertTrue(tracker.addMessage(5, 4, 40)); + + assertFalse(tracker.hasMessageAvailable()); + assertEquals(tracker.getNumberOfDelayedMessages(), 5); + + assertEquals(tracker.getScheduledMessages(10), Collections.emptySet()); + + // Move time forward + clockTime.set(15); + + // Message is rejected by tracker since it's already ready to send + assertFalse(tracker.addMessage(6, 6, 10)); + + assertEquals(tracker.getNumberOfDelayedMessages(), 5); + assertTrue(tracker.hasMessageAvailable()); + Set scheduled = tracker.getScheduledMessages(10); + assertEquals(scheduled.size(), 1); + + // Move time forward + clockTime.set(60); + + assertEquals(tracker.getNumberOfDelayedMessages(), 4); + assertTrue(tracker.hasMessageAvailable()); + scheduled = tracker.getScheduledMessages(1); + assertEquals(scheduled.size(), 1); + + assertEquals(tracker.getNumberOfDelayedMessages(), 3); + assertTrue(tracker.hasMessageAvailable()); + scheduled = tracker.getScheduledMessages(3); + assertEquals(scheduled.size(), 3); + + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + assertFalse(tracker.hasMessageAvailable()); + assertEquals(tracker.getScheduledMessages(10), Collections.emptySet()); + + tracker.close(); + } + + @Test(dataProvider = "delayedTracker") + public void testWithTimer(DelayedDeliveryTracker tracker, NavigableMap tasks) throws Exception { + assertTrue(tasks.isEmpty()); + assertTrue(tracker.addMessage(2, 2, 20)); + assertEquals(tasks.size(), 1); + assertEquals(tasks.firstKey().longValue(), 20); + + assertTrue(tracker.addMessage(1, 1, 10)); + assertEquals(tasks.size(), 1); + assertEquals(tasks.firstKey().longValue(), 10); + + assertTrue(tracker.addMessage(3, 3, 30)); + assertEquals(tasks.size(), 1); + assertEquals(tasks.firstKey().longValue(), 10); + + clockTime.set(15); + + TimerTask task = tasks.pollFirstEntry().getValue(); + Timeout cancelledTimeout = mock(Timeout.class); + when(cancelledTimeout.isCancelled()).thenReturn(true); + task.run(cancelledTimeout); + verify(dispatcher, atMostOnce()).readMoreEntries(); + + task.run(mock(Timeout.class)); + verify(dispatcher).readMoreEntries(); + + tracker.close(); + } + + /** + * Adding a message that is about to expire within the tick time should lead + * to a rejection from the tracker when isDelayedDeliveryDeliverAtTimeStrict is false. + */ + @Test(dataProvider = "delayedTracker") + public void testAddWithinTickTime(DelayedDeliveryTracker tracker) { + clockTime.set(0); + + assertFalse(tracker.addMessage(1, 1, 10)); + assertFalse(tracker.addMessage(2, 2, 99)); + assertFalse(tracker.addMessage(3, 3, 100)); + assertTrue(tracker.addMessage(4, 4, 101)); + assertTrue(tracker.addMessage(5, 5, 200)); + + assertEquals(tracker.getNumberOfDelayedMessages(), 2); + + tracker.close(); + } + + @Test(dataProvider = "delayedTracker") + public void testAddMessageWithStrictDelay(DelayedDeliveryTracker tracker) { + clockTime.set(10); + + // Verify behavior for the less than, equal to, and greater than deliverAt times. + assertFalse(tracker.addMessage(1, 1, 9)); + assertFalse(tracker.addMessage(4, 4, 10)); + assertTrue(tracker.addMessage(1, 1, 11)); + + assertEquals(tracker.getNumberOfDelayedMessages(), 1); + assertFalse(tracker.hasMessageAvailable()); + + tracker.close(); + } + + /** + * In this test, the deliverAt time is after now, but the deliverAt time is too early to run another tick, so the + * tickTimeMillis determines the delay. + */ + @Test(dataProvider = "delayedTracker") + public void testAddMessageWithDeliverAtTimeAfterNowBeforeTickTimeFrequencyWithStrict(DelayedDeliveryTracker tracker) + throws Exception { + // Set clock time, then run tracker to inherit clock time as the last tick time. + clockTime.set(10000); + Timeout timeout = mock(Timeout.class); + when(timeout.isCancelled()).then(x -> false); + ((AbstractDelayedDeliveryTracker) tracker).run(timeout); + verify(dispatcher, times(1)).readMoreEntries(); + + // Add a message that has a delivery time just after the previous run. It will get delivered based on the + // tick delay plus the last tick run. + assertTrue(tracker.addMessage(1, 1, 10001)); + + // Wait longer than the tick time plus the HashedWheelTimer's tick time to ensure that enough time has + // passed where it would have been triggered if the tick time was doing the triggering. + Thread.sleep(600); + verify(dispatcher, times(1)).readMoreEntries(); + + // Not wait for the message delivery to get triggered. + Awaitility.await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> verify(dispatcher).readMoreEntries()); + + tracker.close(); + } + + /** + * In this test, the deliverAt time is after now, but before the (tickTimeMillis + now). Because there wasn't a + * recent tick run, the deliverAt time determines the delay. + */ + @Test(dataProvider = "delayedTracker") + public void testAddMessageWithDeliverAtTimeAfterNowAfterTickTimeFrequencyWithStrict( + DelayedDeliveryTracker tracker) { + clockTime.set(500000); + + assertTrue(tracker.addMessage(1, 1, 500005)); + + // Wait long enough for the runnable to run, but not longer than the tick time. The point is that the delivery + // should get scheduled early when the tick duration has passed since the last tick. + Awaitility.await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> verify(dispatcher).readMoreEntries()); + + tracker.close(); + } + + /** + * In this test, the deliverAt time is after now plus tickTimeMillis, so the tickTimeMillis determines the delay. + */ + @Test(dataProvider = "delayedTracker") + public void testAddMessageWithDeliverAtTimeAfterFullTickTimeWithStrict(DelayedDeliveryTracker tracker) + throws Exception { + clockTime.set(0); + + assertTrue(tracker.addMessage(1, 1, 2000)); + + // Wait longer than the tick time plus the HashedWheelTimer's tick time to ensure that enough time has + // passed where it would have been triggered if the tick time was doing the triggering. + Thread.sleep(1000); + + // Not wait for the message delivery to get triggered. + Awaitility.await().atMost(10, TimeUnit.SECONDS) + .untilAsserted(() -> verify(dispatcher).readMoreEntries()); + + tracker.close(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerTest.java new file mode 100644 index 0000000000000..331ceb83a9994 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerTest.java @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; +import io.netty.util.Timeout; +import io.netty.util.Timer; +import io.netty.util.TimerTask; +import java.lang.reflect.Method; +import java.time.Clock; +import java.util.NavigableMap; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.impl.PositionImpl; +import org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker; +import org.apache.pulsar.broker.delayed.bucket.BucketSnapshotStorage; +import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +@Test(groups = "broker") +public class BucketDelayedDeliveryTrackerTest extends AbstractDeliveryTrackerTest { + + private BucketSnapshotStorage bucketSnapshotStorage; + + @AfterMethod + public void clean() throws Exception { + if (bucketSnapshotStorage != null) { + bucketSnapshotStorage.close(); + } + } + + @DataProvider(name = "delayedTracker") + public Object[][] provider(Method method) throws Exception { + dispatcher = mock(PersistentDispatcherMultipleConsumers.class); + clock = mock(Clock.class); + clockTime = new AtomicLong(); + when(clock.millis()).then(x -> clockTime.get()); + + bucketSnapshotStorage = new MockBucketSnapshotStorage(); + bucketSnapshotStorage.start(); + ManagedCursor cursor = new MockManagedCursor("my_test_cursor"); + doReturn(cursor).when(dispatcher).getCursor(); + + final String methodName = method.getName(); + return switch (methodName) { + case "test" -> new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock, + false, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), 50) + }}; + case "testWithTimer" -> { + Timer timer = mock(Timer.class); + + AtomicLong clockTime = new AtomicLong(); + Clock clock = mock(Clock.class); + when(clock.millis()).then(x -> clockTime.get()); + + NavigableMap tasks = new TreeMap<>(); + + when(timer.newTimeout(any(), anyLong(), any())).then(invocation -> { + TimerTask task = invocation.getArgument(0, TimerTask.class); + long timeout = invocation.getArgument(1, Long.class); + TimeUnit unit = invocation.getArgument(2, TimeUnit.class); + long scheduleAt = clockTime.get() + unit.toMillis(timeout); + tasks.put(scheduleAt, task); + + Timeout t = mock(Timeout.class); + when(t.cancel()).then(i -> { + tasks.remove(scheduleAt, task); + return null; + }); + return t; + }); + + yield new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock, + false, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), 50), + tasks + }}; + } + case "testAddWithinTickTime" -> new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 100, clock, + false, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), 50) + }}; + case "testAddMessageWithStrictDelay" -> new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 100, clock, + true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), 50) + }}; + case "testAddMessageWithDeliverAtTimeAfterNowBeforeTickTimeFrequencyWithStrict" -> new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 1000, clock, + true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), 50) + }}; + case "testAddMessageWithDeliverAtTimeAfterNowAfterTickTimeFrequencyWithStrict", "testRecoverSnapshot" -> + new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 100000, clock, + true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), 50) + }}; + case "testAddMessageWithDeliverAtTimeAfterFullTickTimeWithStrict", "testExistDelayedMessage" -> + new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 500, clock, + true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), 50) + }}; + default -> new Object[][]{{ + new BucketDelayedDeliveryTracker(dispatcher, timer, 1, clock, + true, bucketSnapshotStorage, 1000, TimeUnit.MILLISECONDS.toMillis(100), 50) + }}; + }; + } + + @Test(dataProvider = "delayedTracker") + public void testContainsMessage(DelayedDeliveryTracker tracker) { + tracker.addMessage(1, 1, 10); + tracker.addMessage(2, 2, 20); + + assertTrue(tracker.containsMessage(1, 1)); + clockTime.set(20); + + Set scheduledMessages = tracker.getScheduledMessages(1); + assertEquals(scheduledMessages.stream().findFirst().get().getEntryId(), 1); + + tracker.addMessage(3, 3, 30); + + tracker.addMessage(4, 4, 30); + + tracker.addMessage(5, 5, 30); + + tracker.addMessage(6, 6, 30); + + assertTrue(tracker.containsMessage(3, 3)); + + tracker.close(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java index 1a98233c3859a..6711aed924c20 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/InMemoryDeliveryTrackerTest.java @@ -18,13 +18,9 @@ */ package org.apache.pulsar.broker.delayed; -import static org.mockito.Mockito.any; -import static org.mockito.Mockito.anyLong; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoInteractions; -import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; @@ -35,297 +31,95 @@ import io.netty.util.Timer; import io.netty.util.TimerTask; import io.netty.util.concurrent.DefaultThreadFactory; +import java.lang.reflect.Method; import java.time.Clock; -import java.util.Collections; import java.util.NavigableMap; -import java.util.Set; import java.util.TreeMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; -import lombok.Cleanup; -import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; -import org.awaitility.Awaitility; -import org.testng.annotations.AfterClass; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @Test(groups = "broker") -public class InMemoryDeliveryTrackerTest { +public class InMemoryDeliveryTrackerTest extends AbstractDeliveryTrackerTest { - // Create a single shared timer for the test. - private final Timer timer = new HashedWheelTimer(new DefaultThreadFactory("pulsar-in-memory-delayed-delivery-test"), - 500, TimeUnit.MILLISECONDS); - - @AfterClass(alwaysRun = true) - public void cleanup() { - timer.stop(); - } - - @Test - public void test() throws Exception { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); + @DataProvider(name = "delayedTracker") + public Object[][] provider(Method method) throws Exception { + dispatcher = mock(PersistentDispatcherMultipleConsumers.class); + clock = mock(Clock.class); + clockTime = new AtomicLong(); when(clock.millis()).then(x -> clockTime.get()); - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, - false, 0); - - assertFalse(tracker.hasMessageAvailable()); - - assertTrue(tracker.addMessage(2, 2, 20)); - assertTrue(tracker.addMessage(1, 1, 10)); - assertTrue(tracker.addMessage(3, 3, 30)); - assertTrue(tracker.addMessage(5, 5, 50)); - assertTrue(tracker.addMessage(4, 4, 40)); - - assertFalse(tracker.hasMessageAvailable()); - assertEquals(tracker.getNumberOfDelayedMessages(), 5); - - assertEquals(tracker.getScheduledMessages(10), Collections.emptySet()); - - // Move time forward - clockTime.set(15); - - // Message is rejected by tracker since it's already ready to send - assertFalse(tracker.addMessage(6, 6, 10)); - - assertEquals(tracker.getNumberOfDelayedMessages(), 5); - assertTrue(tracker.hasMessageAvailable()); - Set scheduled = tracker.getScheduledMessages(10); - assertEquals(scheduled.size(), 1); - - // Move time forward - clockTime.set(60); - - assertEquals(tracker.getNumberOfDelayedMessages(), 4); - assertTrue(tracker.hasMessageAvailable()); - scheduled = tracker.getScheduledMessages(1); - assertEquals(scheduled.size(), 1); - - assertEquals(tracker.getNumberOfDelayedMessages(), 3); - assertTrue(tracker.hasMessageAvailable()); - scheduled = tracker.getScheduledMessages(3); - assertEquals(scheduled.size(), 3); - - assertEquals(tracker.getNumberOfDelayedMessages(), 0); - assertFalse(tracker.hasMessageAvailable()); - assertEquals(tracker.getScheduledMessages(10), Collections.emptySet()); - } - - @Test - public void testWithTimer() throws Exception { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - Timer timer = mock(Timer.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - NavigableMap tasks = new TreeMap<>(); - - when(timer.newTimeout(any(), anyLong(), any())).then(invocation -> { - TimerTask task = invocation.getArgument(0, TimerTask.class); - long timeout = invocation.getArgument(1, Long.class); - TimeUnit unit = invocation.getArgument(2, TimeUnit.class); - long scheduleAt = clockTime.get() + unit.toMillis(timeout); - tasks.put(scheduleAt, task); - - Timeout t = mock(Timeout.class); - when(t.cancel()).then(i -> { - tasks.remove(scheduleAt, task); - return null; - }); - return t; - }); - - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, - false, 0); - - assertTrue(tasks.isEmpty()); - assertTrue(tracker.addMessage(2, 2, 20)); - assertEquals(tasks.size(), 1); - assertEquals(tasks.firstKey().longValue(), 20); - - assertTrue(tracker.addMessage(1, 1, 10)); - assertEquals(tasks.size(), 1); - assertEquals(tasks.firstKey().longValue(), 10); - - assertTrue(tracker.addMessage(3, 3, 30)); - assertEquals(tasks.size(), 1); - assertEquals(tasks.firstKey().longValue(), 10); - - clockTime.set(15); - - TimerTask task = tasks.pollFirstEntry().getValue(); - Timeout cancelledTimeout = mock(Timeout.class); - when(cancelledTimeout.isCancelled()).thenReturn(true); - task.run(cancelledTimeout); - verifyZeroInteractions(dispatcher); - - task.run(mock(Timeout.class)); - verify(dispatcher).readMoreEntries(); - } - - /** - * Adding a message that is about to expire within the tick time should lead - * to a rejection from the tracker when isDelayedDeliveryDeliverAtTimeStrict is false. - */ - @Test - public void testAddWithinTickTime() { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, 100, clock, - false, 0); - - clockTime.set(0); - - assertFalse(tracker.addMessage(1, 1, 10)); - assertFalse(tracker.addMessage(2, 2, 99)); - assertFalse(tracker.addMessage(3, 3, 100)); - assertTrue(tracker.addMessage(4, 4, 101)); - assertTrue(tracker.addMessage(5, 5, 200)); - - assertEquals(tracker.getNumberOfDelayedMessages(), 2); - } - - public void testAddMessageWithStrictDelay() { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, 100, clock, - true, 0); - - clockTime.set(10); - - // Verify behavior for the less than, equal to, and greater than deliverAt times. - assertFalse(tracker.addMessage(1, 1, 9)); - assertFalse(tracker.addMessage(4, 4, 10)); - assertTrue(tracker.addMessage(1, 1, 11)); - - assertEquals(tracker.getNumberOfDelayedMessages(), 1); - assertFalse(tracker.hasMessageAvailable()); - } - - /** - * In this test, the deliverAt time is after now, but the deliverAt time is too early to run another tick, so the - * tickTimeMillis determines the delay. - */ - public void testAddMessageWithDeliverAtTimeAfterNowBeforeTickTimeFrequencyWithStrict() throws Exception { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - // Use a short tick time to show that the timer task is run based on the deliverAt time in this scenario. - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, - 1000, clock, true, 0); - - // Set clock time, then run tracker to inherit clock time as the last tick time. - clockTime.set(10000); - Timeout timeout = mock(Timeout.class); - when(timeout.isCancelled()).then(x -> false); - tracker.run(timeout); - verify(dispatcher, times(1)).readMoreEntries(); - - // Add a message that has a delivery time just after the previous run. It will get delivered based on the - // tick delay plus the last tick run. - assertTrue(tracker.addMessage(1, 1, 10001)); - - // Wait longer than the tick time plus the HashedWheelTimer's tick time to ensure that enough time has - // passed where it would have been triggered if the tick time was doing the triggering. - Thread.sleep(600); - verify(dispatcher, times(1)).readMoreEntries(); - - // Not wait for the message delivery to get triggered. - Awaitility.await().atMost(10, TimeUnit.SECONDS) - .untilAsserted(() -> verify(dispatcher).readMoreEntries()); - } - - /** - * In this test, the deliverAt time is after now, but before the (tickTimeMillis + now). Because there wasn't a - * recent tick run, the deliverAt time determines the delay. - */ - public void testAddMessageWithDeliverAtTimeAfterNowAfterTickTimeFrequencyWithStrict() { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - // Use a large tick time to show that the message will get delivered earlier because there wasn't - // a previous tick run. - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, - 100000, clock, true, 0); - - clockTime.set(500000); - - assertTrue(tracker.addMessage(1, 1, 500005)); - - // Wait long enough for the runnable to run, but not longer than the tick time. The point is that the delivery - // should get scheduled early when the tick duration has passed since the last tick. - Awaitility.await().atMost(10, TimeUnit.SECONDS) - .untilAsserted(() -> verify(dispatcher).readMoreEntries()); - } - - /** - * In this test, the deliverAt time is after now plus tickTimeMillis, so the tickTimeMillis determines the delay. - */ - public void testAddMessageWithDeliverAtTimeAfterFullTickTimeWithStrict() throws Exception { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - // Use a short tick time to show that the timer task is run based on the deliverAt time in this scenario. - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, - 500, clock, true, 0); - - clockTime.set(0); - - assertTrue(tracker.addMessage(1, 1, 2000)); - - // Wait longer than the tick time plus the HashedWheelTimer's tick time to ensure that enough time has - // passed where it would have been triggered if the tick time was doing the triggering. - Thread.sleep(1000); - verifyNoInteractions(dispatcher); - - // Not wait for the message delivery to get triggered. - Awaitility.await().atMost(10, TimeUnit.SECONDS) - .untilAsserted(() -> verify(dispatcher).readMoreEntries()); + final String methodName = method.getName(); + return switch (methodName) { + case "test" -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, + false, 0) + }}; + case "testWithTimer" -> { + Timer timer = mock(Timer.class); + + AtomicLong clockTime = new AtomicLong(); + Clock clock = mock(Clock.class); + when(clock.millis()).then(x -> clockTime.get()); + + NavigableMap tasks = new TreeMap<>(); + + when(timer.newTimeout(any(), anyLong(), any())).then(invocation -> { + TimerTask task = invocation.getArgument(0, TimerTask.class); + long timeout = invocation.getArgument(1, Long.class); + TimeUnit unit = invocation.getArgument(2, TimeUnit.class); + long scheduleAt = clockTime.get() + unit.toMillis(timeout); + tasks.put(scheduleAt, task); + + Timeout t = mock(Timeout.class); + when(t.cancel()).then(i -> { + tasks.remove(scheduleAt, task); + return null; + }); + return t; + }); + + yield new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, + false, 0), + tasks + }}; + } + case "testAddWithinTickTime" -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 100, clock, + false, 0) + }}; + case "testAddMessageWithStrictDelay" -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 100, clock, + true, 0) + }}; + case "testAddMessageWithDeliverAtTimeAfterNowBeforeTickTimeFrequencyWithStrict" -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1000, clock, + true, 0) + }}; + case "testAddMessageWithDeliverAtTimeAfterNowAfterTickTimeFrequencyWithStrict" -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 100000, clock, + true, 0) + }}; + case "testAddMessageWithDeliverAtTimeAfterFullTickTimeWithStrict" -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 500, clock, + true, 0) + }}; + case "testWithFixedDelays", "testWithMixedDelays","testWithNoDelays" -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 500, clock, + true, 100) + }}; + default -> new Object[][]{{ + new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, + true, 0) + }}; + }; } - @Test - public void testWithFixedDelays() throws Exception { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - final long fixedDelayLookahead = 100; - - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, - true, fixedDelayLookahead); - + @Test(dataProvider = "delayedTracker") + public void testWithFixedDelays(InMemoryDelayedDeliveryTracker tracker) throws Exception { assertFalse(tracker.hasMessageAvailable()); assertTrue(tracker.addMessage(1, 1, 10)); @@ -338,15 +132,16 @@ public void testWithFixedDelays() throws Exception { assertEquals(tracker.getNumberOfDelayedMessages(), 5); assertFalse(tracker.shouldPauseAllDeliveries()); - for (int i = 6; i <= fixedDelayLookahead; i++) { + for (int i = 6; i <= tracker.getFixedDelayDetectionLookahead(); i++) { assertTrue(tracker.addMessage(i, i, i * 10)); } assertTrue(tracker.shouldPauseAllDeliveries()); - clockTime.set(fixedDelayLookahead * 10); + clockTime.set(tracker.getFixedDelayDetectionLookahead() * 10); tracker.getScheduledMessages(100); + assertFalse(tracker.shouldPauseAllDeliveries()); // Empty the tracker @@ -356,22 +151,12 @@ public void testWithFixedDelays() throws Exception { } while (removed > 0); assertFalse(tracker.shouldPauseAllDeliveries()); - } - - @Test - public void testWithMixedDelays() throws Exception { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - long fixedDelayLookahead = 100; - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, - true, fixedDelayLookahead); + tracker.close(); + } + @Test(dataProvider = "delayedTracker") + public void testWithMixedDelays(InMemoryDelayedDeliveryTracker tracker) throws Exception { assertFalse(tracker.hasMessageAvailable()); assertTrue(tracker.addMessage(1, 1, 10)); @@ -382,32 +167,22 @@ public void testWithMixedDelays() throws Exception { assertFalse(tracker.shouldPauseAllDeliveries()); - for (int i = 6; i <= fixedDelayLookahead; i++) { + for (int i = 6; i <= tracker.getFixedDelayDetectionLookahead(); i++) { assertTrue(tracker.addMessage(i, i, i * 10)); } assertTrue(tracker.shouldPauseAllDeliveries()); // Add message with earlier delivery time - assertTrue(tracker.addMessage(5, 5, 5)); + assertTrue(tracker.addMessage(5, 6, 5)); assertFalse(tracker.shouldPauseAllDeliveries()); - } - - @Test - public void testWithNoDelays() throws Exception { - PersistentDispatcherMultipleConsumers dispatcher = mock(PersistentDispatcherMultipleConsumers.class); - - AtomicLong clockTime = new AtomicLong(); - Clock clock = mock(Clock.class); - when(clock.millis()).then(x -> clockTime.get()); - - long fixedDelayLookahead = 100; - @Cleanup - InMemoryDelayedDeliveryTracker tracker = new InMemoryDelayedDeliveryTracker(dispatcher, timer, 1, clock, - true, fixedDelayLookahead); + tracker.close(); + } + @Test(dataProvider = "delayedTracker") + public void testWithNoDelays(InMemoryDelayedDeliveryTracker tracker) throws Exception { assertFalse(tracker.hasMessageAvailable()); assertTrue(tracker.addMessage(1, 1, 10)); @@ -418,16 +193,18 @@ public void testWithNoDelays() throws Exception { assertFalse(tracker.shouldPauseAllDeliveries()); - for (int i = 6; i <= fixedDelayLookahead; i++) { + for (int i = 6; i <= tracker.getFixedDelayDetectionLookahead(); i++) { assertTrue(tracker.addMessage(i, i, i * 10)); } assertTrue(tracker.shouldPauseAllDeliveries()); // Add message with no-delay - assertFalse(tracker.addMessage(5, 5, -1L)); + assertFalse(tracker.addMessage(5, 6, -1L)); assertFalse(tracker.shouldPauseAllDeliveries()); + + tracker.close(); } @Test @@ -471,5 +248,4 @@ public void run(Timeout timeout) throws Exception { timer.stop(); } - } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java new file mode 100644 index 0000000000000..89831a1d5e771 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java @@ -0,0 +1,160 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed; + +import com.google.protobuf.InvalidProtocolBufferException; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.PooledByteBufAllocator; +import io.netty.util.concurrent.DefaultThreadFactory; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.broker.delayed.bucket.BucketSnapshotStorage; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.SnapshotMetadata; +import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment; + +@Slf4j +public class MockBucketSnapshotStorage implements BucketSnapshotStorage { + + private final AtomicLong maxBucketId; + + private final Map> bucketSnapshots; + + private final ExecutorService executorService = + new ThreadPoolExecutor(10, 20, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), + new DefaultThreadFactory("bucket-snapshot-storage-io")); + + public MockBucketSnapshotStorage() { + this.bucketSnapshots = new ConcurrentHashMap<>(); + this.maxBucketId = new AtomicLong(); + } + + @Override + public CompletableFuture createBucketSnapshot( + SnapshotMetadata snapshotMetadata, List bucketSnapshotSegments) { + return CompletableFuture.supplyAsync(() -> { + long bucketId = maxBucketId.getAndIncrement(); + List entries = new ArrayList<>(); + byte[] bytes = snapshotMetadata.toByteArray(); + ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(bytes.length); + byteBuf.writeBytes(bytes); + entries.add(byteBuf); + this.bucketSnapshots.put(bucketId, entries); + return bucketId; + }, executorService).thenApply(bucketId -> { + List bufList = new ArrayList<>(); + for (SnapshotSegment snapshotSegment : bucketSnapshotSegments) { + byte[] bytes = snapshotSegment.toByteArray(); + ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(bytes.length); + byteBuf.writeBytes(bytes); + bufList.add(byteBuf); + } + bucketSnapshots.get(bucketId).addAll(bufList); + + return bucketId; + }); + } + + @Override + public CompletableFuture getBucketSnapshotMetadata(long bucketId) { + return CompletableFuture.supplyAsync(() -> { + ByteBuf byteBuf = this.bucketSnapshots.get(bucketId).get(0); + SnapshotMetadata snapshotMetadata; + try { + snapshotMetadata = SnapshotMetadata.parseFrom(byteBuf.nioBuffer()); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + return snapshotMetadata; + }, executorService); + } + + @Override + public CompletableFuture> getBucketSnapshotSegment(long bucketId, long firstSegmentEntryId, + long lastSegmentEntryId) { + return CompletableFuture.supplyAsync(() -> { + List snapshotSegments = new ArrayList<>(); + long lastEntryId = Math.min(lastSegmentEntryId, this.bucketSnapshots.get(bucketId).size()); + for (int i = (int) firstSegmentEntryId; i <= lastEntryId ; i++) { + ByteBuf byteBuf = this.bucketSnapshots.get(bucketId).get(i); + SnapshotSegment snapshotSegment; + try { + snapshotSegment = SnapshotSegment.parseFrom(byteBuf.nioBuffer()); + snapshotSegments.add(snapshotSegment); + } catch (InvalidProtocolBufferException e) { + throw new RuntimeException(e); + } + } + return snapshotSegments; + }, executorService); + } + + @Override + public CompletableFuture deleteBucketSnapshot(long bucketId) { + return CompletableFuture.supplyAsync(() -> { + List remove = this.bucketSnapshots.remove(bucketId); + if (remove != null) { + for (ByteBuf byteBuf : remove) { + byteBuf.release(); + } + } + return null; + }, executorService); + } + + @Override + public CompletableFuture getBucketSnapshotLength(long bucketId) { + return CompletableFuture.supplyAsync(() -> { + long length = 0; + List bufList = this.bucketSnapshots.get(bucketId); + for (ByteBuf byteBuf : bufList) { + length += byteBuf.array().length; + } + return length; + }, executorService); + } + + @Override + public void start() throws Exception { + + } + + @Override + public void close() throws Exception { + clean(); + } + + public void clean() { + for (List value : bucketSnapshots.values()) { + for (ByteBuf byteBuf : value) { + byteBuf.release(); + } + } + bucketSnapshots.clear(); + executorService.shutdownNow(); + } +} diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockManagedCursor.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockManagedCursor.java new file mode 100644 index 0000000000000..efb0fa7ab7ba2 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockManagedCursor.java @@ -0,0 +1,412 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.delayed; + +import com.google.common.collect.Range; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.bookkeeper.mledger.AsyncCallbacks; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.ManagedCursorMXBean; +import org.apache.bookkeeper.mledger.ManagedLedger; +import org.apache.bookkeeper.mledger.ManagedLedgerException; +import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.impl.PositionImpl; + +public class MockManagedCursor implements ManagedCursor { + + private final String name; + + private final Map cursorProperties; + + public MockManagedCursor(String name) { + this.name = name; + this.cursorProperties = new ConcurrentHashMap<>(); + } + + @Override + public String getName() { + return null; + } + + @Override + public long getLastActive() { + return 0; + } + + @Override + public void updateLastActive() { + + } + + @Override + public Map getProperties() { + return null; + } + + @Override + public Map getCursorProperties() { + return this.cursorProperties; + } + + @Override + public CompletableFuture putCursorProperty(String key, String value) { + cursorProperties.put(key, value); + return CompletableFuture.completedFuture(null); + } + + @Override + public CompletableFuture setCursorProperties(Map cursorProperties) { + return CompletableFuture.completedFuture(null); + } + + public CompletableFuture removeCursorProperty(String key) { + cursorProperties.remove(key); + return CompletableFuture.completedFuture(null); + } + + @Override + public boolean putProperty(String key, Long value) { + return false; + } + + @Override + public boolean removeProperty(String key) { + return false; + } + + @Override + public List readEntries(int numberOfEntriesToRead) throws InterruptedException, ManagedLedgerException { + return null; + } + + @Override + public void asyncReadEntries(int numberOfEntriesToRead, AsyncCallbacks.ReadEntriesCallback callback, Object ctx, + PositionImpl maxPosition) { + + } + + @Override + public void asyncReadEntries(int numberOfEntriesToRead, long maxSizeBytes, + AsyncCallbacks.ReadEntriesCallback callback, Object ctx, PositionImpl maxPosition) { + + } + + @Override + public Entry getNthEntry(int n, IndividualDeletedEntries deletedEntries) + throws InterruptedException, ManagedLedgerException { + return null; + } + + @Override + public void asyncGetNthEntry(int n, IndividualDeletedEntries deletedEntries, + AsyncCallbacks.ReadEntryCallback callback, Object ctx) { + + } + + @Override + public List readEntriesOrWait(int numberOfEntriesToRead) + throws InterruptedException, ManagedLedgerException { + return null; + } + + @Override + public List readEntriesOrWait(int maxEntries, long maxSizeBytes) + throws InterruptedException, ManagedLedgerException { + return null; + } + + @Override + public void asyncReadEntriesOrWait(int numberOfEntriesToRead, AsyncCallbacks.ReadEntriesCallback callback, + Object ctx, PositionImpl maxPosition) { + + } + + @Override + public void asyncReadEntriesOrWait(int maxEntries, long maxSizeBytes, AsyncCallbacks.ReadEntriesCallback callback, + Object ctx, PositionImpl maxPosition) { + + } + + @Override + public boolean cancelPendingReadRequest() { + return false; + } + + @Override + public boolean hasMoreEntries() { + return false; + } + + @Override + public long getNumberOfEntries() { + return 0; + } + + @Override + public long getNumberOfEntriesInBacklog(boolean isPrecise) { + return 0; + } + + @Override + public void markDelete(Position position) throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void markDelete(Position position, Map properties) + throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void asyncMarkDelete(Position position, AsyncCallbacks.MarkDeleteCallback callback, Object ctx) { + + } + + @Override + public void asyncMarkDelete(Position position, Map properties, + AsyncCallbacks.MarkDeleteCallback callback, Object ctx) { + + } + + @Override + public void delete(Position position) throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void asyncDelete(Position position, AsyncCallbacks.DeleteCallback callback, Object ctx) { + + } + + @Override + public void delete(Iterable positions) throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void asyncDelete(Iterable position, AsyncCallbacks.DeleteCallback callback, Object ctx) { + + } + + @Override + public Position getReadPosition() { + return null; + } + + @Override + public Position getMarkDeletedPosition() { + return null; + } + + @Override + public Position getPersistentMarkDeletedPosition() { + return null; + } + + @Override + public void rewind() { + + } + + @Override + public void seek(Position newReadPosition, boolean force) { + + } + + @Override + public void clearBacklog() throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void asyncClearBacklog(AsyncCallbacks.ClearBacklogCallback callback, Object ctx) { + + } + + @Override + public void skipEntries(int numEntriesToSkip, IndividualDeletedEntries deletedEntries) + throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void asyncSkipEntries(int numEntriesToSkip, IndividualDeletedEntries deletedEntries, + AsyncCallbacks.SkipEntriesCallback callback, Object ctx) { + + } + + @Override + public Position findNewestMatching(java.util.function.Predicate condition) + throws InterruptedException, ManagedLedgerException { + return null; + } + + @Override + public Position findNewestMatching(FindPositionConstraint constraint, java.util.function.Predicate condition) + throws InterruptedException, ManagedLedgerException { + return null; + } + + @Override + public void asyncFindNewestMatching(FindPositionConstraint constraint, + java.util.function.Predicate condition, + AsyncCallbacks.FindEntryCallback callback, Object ctx) { + + } + + @Override + public void resetCursor(Position position) throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void asyncResetCursor(Position position, boolean forceReset, AsyncCallbacks.ResetCursorCallback callback) { + + } + + @Override + public List replayEntries(Set positions) + throws InterruptedException, ManagedLedgerException { + return null; + } + + @Override + public Set asyncReplayEntries(Set positions, + AsyncCallbacks.ReadEntriesCallback callback, Object ctx) { + return null; + } + + @Override + public Set asyncReplayEntries(Set positions, + AsyncCallbacks.ReadEntriesCallback callback, Object ctx, + boolean sortEntries) { + return null; + } + + @Override + public void close() throws InterruptedException, ManagedLedgerException { + + } + + @Override + public void asyncClose(AsyncCallbacks.CloseCallback callback, Object ctx) { + + } + + @Override + public Position getFirstPosition() { + return null; + } + + @Override + public void setActive() { + + } + + @Override + public void setInactive() { + + } + + @Override + public void setAlwaysInactive() { + + } + + @Override + public boolean isActive() { + return false; + } + + @Override + public boolean isDurable() { + return false; + } + + @Override + public long getNumberOfEntriesSinceFirstNotAckedMessage() { + return 0; + } + + @Override + public int getTotalNonContiguousDeletedMessagesRange() { + return 0; + } + + @Override + public int getNonContiguousDeletedMessagesRangeSerializedSize() { + return 0; + } + + @Override + public long getEstimatedSizeSinceMarkDeletePosition() { + return 0; + } + + @Override + public double getThrottleMarkDelete() { + return 0; + } + + @Override + public void setThrottleMarkDelete(double throttleMarkDelete) { + + } + + @Override + public ManagedLedger getManagedLedger() { + return null; + } + + @Override + public Range getLastIndividualDeletedRange() { + return null; + } + + @Override + public void trimDeletedEntries(List entries) { + + } + + @Override + public long[] getDeletedBatchIndexesAsLongArray(PositionImpl position) { + return new long[0]; + } + + @Override + public ManagedCursorMXBean getStats() { + return null; + } + + @Override + public boolean checkAndUpdateReadPositionChanged() { + return false; + } + + @Override + public boolean isClosed() { + return false; + } +}