diff --git a/microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java index 08d02195c8781..a8c1019f02215 100644 --- a/microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java +++ b/microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerBenchmark.java @@ -53,15 +53,18 @@ *

This benchmark measures tracker throughput under different read/write ratios * and initial message counts without implying a specific lock implementation. * - *

Run with: mvn exec:java -Dexec.mainClass="org.openjdk.jmh.Main" - * -Dexec.args="BucketDelayedDeliveryTrackerBenchmark" + *

Run with: + * 

+ *   ./gradlew :microbench:shadowJar
+ *   java -jar microbench/build/libs/microbench-*-benchmarks.jar BucketDelayedDeliveryTrackerBenchmark
+ *
*/ @BenchmarkMode(Mode.Throughput) @OutputTimeUnit(TimeUnit.SECONDS) @State(Scope.Benchmark) -@Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1) -@Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 1) -@Fork(1) +@Warmup(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 3) +@Measurement(time = 10, timeUnit = TimeUnit.SECONDS, iterations = 5) +@Fork(2) public class BucketDelayedDeliveryTrackerBenchmark { /** diff --git a/microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerSealBenchmark.java b/microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerSealBenchmark.java new file mode 100644 index 0000000000000..0131f9e7675be --- /dev/null +++ b/microbench/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerSealBenchmark.java @@ -0,0 +1,141 @@ +/* + * 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 io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import io.netty.util.concurrent.DefaultThreadFactory; +import java.time.Clock; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.mledger.PositionFactory; +import org.apache.bookkeeper.mledger.impl.ActiveManagedCursorContainerImpl; +import org.apache.bookkeeper.mledger.impl.MockManagedCursor; +import org.apache.pulsar.broker.delayed.NoopDelayedDeliveryContext; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; + +/** + * Measures the latency of the add that rolls a full mutable bucket to a new ledger. + * + *

The invocation setup preloads the full bucket outside the measured interval. The direct mode is a + * same-code blocking baseline; the async mode measures the seal-and-swap path used by the broker factory. + * + *

Run with: + * 

+ *   ./gradlew :microbench:shadowJar
+ *   java -jar microbench/build/libs/microbench-*-benchmarks.jar \
+ *       BucketDelayedDeliveryTrackerSealBenchmark
+ *
+ */ +@BenchmarkMode(Mode.SingleShotTime) +@OutputTimeUnit(TimeUnit.MICROSECONDS) +@Warmup(iterations = 3) +@Measurement(iterations = 5) +@Fork(2) +@State(Scope.Thread) +public class BucketDelayedDeliveryTrackerSealBenchmark { + + private static final long FUTURE_DELIVERY_BASE_TIME_MILLIS = 4102444800000L; + + @Param({"10000", "100000", "1000000"}) + public int indexesPerBucket; + + @Param({"direct", "async"}) + public String buildMode; + + private BucketDelayedDeliveryTracker tracker; + private Timer timer; + private MockBucketSnapshotStorage storage; + private ExecutorService snapshotBuildExecutor; + + @Setup(Level.Invocation) + public void setupInvocation() throws Exception { + timer = new HashedWheelTimer(new DefaultThreadFactory("seal-rollover-timer"), + 100, TimeUnit.MILLISECONDS); + storage = new MockBucketSnapshotStorage(); + storage.start(); + + ActiveManagedCursorContainerImpl container = new ActiveManagedCursorContainerImpl(); + MockManagedCursor cursor = MockManagedCursor.createCursor(container, "seal-rollover-cursor", + PositionFactory.create(0, 0)); + String dispatcherName = "persistent://public/default/seal-rollover / " + cursor.getName(); + NoopDelayedDeliveryContext context = new NoopDelayedDeliveryContext(dispatcherName, cursor); + + Executor executor; + snapshotBuildExecutor = null; + if ("async".equals(buildMode)) { + snapshotBuildExecutor = Executors.newSingleThreadExecutor( + new DefaultThreadFactory("seal-rollover-snapshot-builder")); + executor = snapshotBuildExecutor; + } else { + executor = Runnable::run; + } + tracker = new BucketDelayedDeliveryTracker(context, timer, 1_000, Clock.systemUTC(), true, storage, + indexesPerBucket, TimeUnit.MINUTES.toMillis(5), 5_000, -1, executor); + + for (int entryId = 0; entryId < indexesPerBucket; entryId++) { + tracker.addMessage(1, entryId, FUTURE_DELIVERY_BASE_TIME_MILLIS + entryId); + } + } + + @TearDown(Level.Invocation) + public void tearDownInvocation() throws Exception { + if (tracker != null) { + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (tracker.getBucketsCount().get() == 0 && System.nanoTime() < deadlineNanos) { + Thread.sleep(1); + } + if (tracker.getBucketsCount().get() != 1) { + throw new IllegalStateException("The rollover snapshot did not commit"); + } + tracker.close(); + } + if (snapshotBuildExecutor != null) { + snapshotBuildExecutor.shutdown(); + if (!snapshotBuildExecutor.awaitTermination(30, TimeUnit.SECONDS)) { + snapshotBuildExecutor.shutdownNow(); + } + } + if (storage != null) { + storage.close(); + } + if (timer != null) { + timer.stop(); + } + } + + @Benchmark + public boolean sealRolloverAddLatency() { + return tracker.addMessage(2, 0, FUTURE_DELIVERY_BASE_TIME_MILLIS); + } +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java index 80d1076c1c8f4..dcc4e1c7c2035 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java @@ -25,7 +25,11 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import lombok.CustomLog; import org.apache.bookkeeper.mledger.ManagedCursor; @@ -39,6 +43,7 @@ import org.apache.pulsar.broker.service.BrokerService; import org.apache.pulsar.broker.service.persistent.AbstractPersistentDispatcherMultipleConsumers; import org.apache.pulsar.common.util.FutureUtil; +import org.apache.pulsar.common.util.GracefulExecutorServicesShutdown; @CustomLog public class BucketDelayedDeliveryTrackerFactory implements DelayedDeliveryTrackerFactory { @@ -58,6 +63,8 @@ public class BucketDelayedDeliveryTrackerFactory implements DelayedDeliveryTrack private int delayedDeliveryMaxIndexesPerBucketSnapshotSegment; + private ExecutorService snapshotBuildExecutor; + @Override public void initialize(PulsarService pulsarService) throws Exception { ServiceConfiguration config = pulsarService.getConfig(); @@ -73,6 +80,14 @@ public void initialize(PulsarService pulsarService) throws Exception { config.getDelayedDeliveryMaxTimeStepPerBucketSnapshotSegmentSeconds(); this.delayedDeliveryMaxIndexesPerBucketSnapshotSegment = config.getDelayedDeliveryMaxIndexesPerBucketSnapshotSegment(); + int snapshotBuildThreads = Math.max(1, config.getNumExecutorThreadPoolSize()); + // Don't queue detached buckets: they retain off-heap indexes and temporarily delay their delivery. + // Saturated submissions are rejected so the tracker can immediately fall back to shared memory mode. + BlockingQueue workQueue = new SynchronousQueue<>(); + this.snapshotBuildExecutor = new ThreadPoolExecutor(snapshotBuildThreads, snapshotBuildThreads, + 0L, TimeUnit.MILLISECONDS, workQueue, + new DefaultThreadFactory("pulsar-delayed-delivery-snapshot-builder"), + new ThreadPoolExecutor.AbortPolicy()); } @Override @@ -104,7 +119,8 @@ BucketDelayedDeliveryTracker newTracker0(AbstractPersistentDispatcherMultipleCon return new BucketDelayedDeliveryTracker(dispatcher, timer, tickTimeMillis, isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, delayedDeliveryMinIndexCountPerBucket, TimeUnit.SECONDS.toMillis(delayedDeliveryMaxTimeStepPerBucketSnapshotSegmentSeconds), - delayedDeliveryMaxIndexesPerBucketSnapshotSegment, delayedDeliveryMaxNumBuckets); + delayedDeliveryMaxIndexesPerBucketSnapshotSegment, delayedDeliveryMaxNumBuckets, + snapshotBuildExecutor); } /** @@ -133,6 +149,9 @@ public CompletableFuture cleanResidualSnapshots(ManagedCursor cursor) { @Override public void close() throws Exception { + if (snapshotBuildExecutor != null) { + GracefulExecutorServicesShutdown.initiate().shutdown(snapshotBuildExecutor).handle().join(); + } if (bucketSnapshotStorage != null) { bucketSnapshotStorage.close(); } 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 index bd7c596561cc5..55871b3fefb1b 100644 --- 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 @@ -42,6 +42,8 @@ import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; @@ -88,6 +90,8 @@ public static record SnapshotKey(long ledgerId, long entryId) {} private static final int MAX_MERGE_NUM = 4; + private static final Executor DIRECT_EXECUTOR = Runnable::run; + private final long minIndexCountPerBucket; private final long timeStepPerBucketSnapshotSegmentInMillis; @@ -102,7 +106,7 @@ public static record SnapshotKey(long ledgerId, long entryId) {} @Getter @VisibleForTesting - private final MutableBucket lastMutableBucket; + private volatile MutableBucket lastMutableBucket; @Getter @VisibleForTesting @@ -128,6 +132,22 @@ public static record SnapshotKey(long ledgerId, long entryId) {} private final BucketDelayedMessageIndexStats stats; + private final Executor snapshotBuildExecutor; + + private final AtomicLong pendingSealMemoryUsage = new AtomicLong(); + + private final AtomicLong pendingSealMessageCount = new AtomicLong(); + + private PendingSeal pendingSeal; + + private long lifecycleGeneration; + + private Lifecycle lifecycle = Lifecycle.OPEN; + + private CompletableFuture clearFuture; + + private CompletableFuture closeFuture; + private CompletableFuture pendingLoad = null; private volatile CompletableFuture trimFuture; @@ -141,7 +161,22 @@ public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumer throws RecoverDelayedDeliveryTrackerException { this(new DispatcherDelayedDeliveryContext(dispatcher), timer, tickTimeMillis, Clock.systemUTC(), isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, minIndexCountPerBucket, - timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets); + timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets, + DIRECT_EXECUTOR); + } + + public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, + Timer timer, long tickTimeMillis, + boolean isDelayedDeliveryDeliverAtTimeStrict, + BucketSnapshotStorage bucketSnapshotStorage, + long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegmentInMillis, + int maxIndexesPerBucketSnapshotSegment, int maxNumBuckets, + Executor snapshotBuildExecutor) + throws RecoverDelayedDeliveryTrackerException { + this(new DispatcherDelayedDeliveryContext(dispatcher), timer, tickTimeMillis, Clock.systemUTC(), + isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, minIndexCountPerBucket, + timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets, + snapshotBuildExecutor); } public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, @@ -153,7 +188,23 @@ public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumer throws RecoverDelayedDeliveryTrackerException { this(new DispatcherDelayedDeliveryContext(dispatcher), timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, minIndexCountPerBucket, - timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets); + timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets, + DIRECT_EXECUTOR); + } + + @VisibleForTesting + public BucketDelayedDeliveryTracker(AbstractPersistentDispatcherMultipleConsumers dispatcher, + Timer timer, long tickTimeMillis, Clock clock, + boolean isDelayedDeliveryDeliverAtTimeStrict, + BucketSnapshotStorage bucketSnapshotStorage, + long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegmentInMillis, + int maxIndexesPerBucketSnapshotSegment, int maxNumBuckets, + Executor snapshotBuildExecutor) + throws RecoverDelayedDeliveryTrackerException { + this(new DispatcherDelayedDeliveryContext(dispatcher), timer, tickTimeMillis, clock, + isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, minIndexCountPerBucket, + timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, maxNumBuckets, + snapshotBuildExecutor); } @VisibleForTesting @@ -164,6 +215,20 @@ public BucketDelayedDeliveryTracker(DelayedDeliveryContext context, long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegmentInMillis, int maxIndexesPerBucketSnapshotSegment, int maxNumBuckets) throws RecoverDelayedDeliveryTrackerException { + this(context, timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict, bucketSnapshotStorage, + minIndexCountPerBucket, timeStepPerBucketSnapshotSegmentInMillis, + maxIndexesPerBucketSnapshotSegment, maxNumBuckets, DIRECT_EXECUTOR); + } + + @VisibleForTesting + public BucketDelayedDeliveryTracker(DelayedDeliveryContext context, + Timer timer, long tickTimeMillis, Clock clock, + boolean isDelayedDeliveryDeliverAtTimeStrict, + BucketSnapshotStorage bucketSnapshotStorage, + long minIndexCountPerBucket, long timeStepPerBucketSnapshotSegmentInMillis, + int maxIndexesPerBucketSnapshotSegment, int maxNumBuckets, + Executor snapshotBuildExecutor) + throws RecoverDelayedDeliveryTrackerException { super(context, timer, tickTimeMillis, clock, isDelayedDeliveryDeliverAtTimeStrict); this.log = LOG.with().ctx(super.log).build(); this.minIndexCountPerBucket = minIndexCountPerBucket; @@ -177,6 +242,7 @@ public BucketDelayedDeliveryTracker(DelayedDeliveryContext context, bucketSnapshotStorage); this.lastMutableBucket = new MutableBucket(ctx); this.stats = new BucketDelayedMessageIndexStats(); + this.snapshotBuildExecutor = snapshotBuildExecutor; // Close the tracker if failed to recover. try { @@ -187,6 +253,17 @@ public BucketDelayedDeliveryTracker(DelayedDeliveryContext context, } } + private enum Lifecycle { + OPEN, + CLEARING, + CLOSING, + CLOSED + } + + private record PendingSeal(long generation, MutableBucket bucket, long nextDeliveryTime, + long startTime, CompletableFuture completion) { + } + private synchronized long recoverBucketSnapshot() throws RecoverDelayedDeliveryTrackerException { ManagedCursor cursor = ctx.cursor(); Map cursorProperties = cursor.getCursorProperties(); @@ -329,7 +406,7 @@ private synchronized void putAndCleanOverlapRange(Range range, ImmutableBu @Override public void run(Timeout timeout) throws Exception { synchronized (this) { - if (timeout == null || timeout.isCancelled()) { + if (lifecycle != Lifecycle.OPEN || timeout == null || timeout.isCancelled()) { return; } lastMutableBucket.moveScheduledMessageToSharedQueue(getCutoffTime(), sharedBucketPriorityQueue); @@ -382,10 +459,12 @@ private void afterCreateImmutableBucket(Pair immu // Put indexes back into the shared queue and downgrade to memory mode synchronized (BucketDelayedDeliveryTracker.this) { immutableBucket.getSnapshotSegments().ifPresent(snapshotSegments -> { - for (SnapshotSegment snapshotSegment : snapshotSegments) { - for (DelayedIndex delayedIndex : snapshotSegment.getIndexesList()) { - sharedBucketPriorityQueue.add(delayedIndex.getTimestamp(), - delayedIndex.getLedgerId(), delayedIndex.getEntryId()); + if (lifecycle == Lifecycle.OPEN) { + for (SnapshotSegment snapshotSegment : snapshotSegments) { + for (DelayedIndex delayedIndex : snapshotSegment.getIndexesList()) { + sharedBucketPriorityQueue.add(delayedIndex.getTimestamp(), + delayedIndex.getLedgerId(), delayedIndex.getEntryId()); + } } } immutableBucket.setSnapshotSegments(null); @@ -405,62 +484,245 @@ private void afterCreateImmutableBucket(Pair immu } @Override - public synchronized boolean addMessage(long ledgerId, long entryId, long deliverAt) { - if (deliverAt < 0 || deliverAt <= getCutoffTime()) { - removeIndexBit(ledgerId, entryId); - return false; - } + public boolean addMessage(long ledgerId, long entryId, long deliverAt) { + PendingSeal sealToSubmit = null; + synchronized (this) { + if (lifecycle != Lifecycle.OPEN) { + return false; + } - if (containsMessage(ledgerId, entryId)) { - return true; - } + if (deliverAt < 0 || deliverAt <= getCutoffTime()) { + removeIndexBit(ledgerId, entryId); + return false; + } - boolean existBucket = findImmutableBucket(ledgerId) != null; - - // Create bucket snapshot - if (!existBucket && ledgerId > lastMutableBucket.endLedgerId - && lastMutableBucket.size() >= minIndexCountPerBucket - && !lastMutableBucket.isEmpty()) { - long createStartTime = System.currentTimeMillis(); - stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.create); - Pair immutableBucketDelayedIndexPair = - lastMutableBucket.sealBucketAndAsyncPersistent( - this.timeStepPerBucketSnapshotSegmentInMillis, - this.maxIndexesPerBucketSnapshotSegment, - this.sharedBucketPriorityQueue); - afterCreateImmutableBucket(immutableBucketDelayedIndexPair, createStartTime); - lastMutableBucket.resetLastMutableBucketRange(); - - if (maxNumBuckets > 0 && bucketsCount.get() > maxNumBuckets - && (trimFuture == null || trimFuture.isDone())) { - trimFuture = asyncTrimImmutableBuckets() - .thenCompose(ignore -> asyncMergeBucketSnapshot()) - .whenComplete((ignore, t) -> { - if (t != null) { - log.warn().exception(t).log("Failed to trim or merge bucket snapshots"); - } - }); + if (index.contains(ledgerId, entryId)) { + return true; } - } - if (ledgerId >= lastMutableBucket.endLedgerId && !existBucket) { - lastMutableBucket.addMessage(ledgerId, entryId, deliverAt); - } else { - // Message index belongs to previous bucket range or the current mutable bucket range, - // enter sharedBucketPriorityQueue directly - sharedBucketPriorityQueue.add(deliverAt, ledgerId, entryId); - } - index.track(ledgerId, entryId); + boolean existBucket = findImmutableBucket(ledgerId) != null; + if (pendingSeal == null && !existBucket && ledgerId > lastMutableBucket.endLedgerId + && lastMutableBucket.size() >= minIndexCountPerBucket + && !lastMutableBucket.isEmpty()) { + MutableBucket detachedBucket = lastMutableBucket; + lastMutableBucket = new MutableBucket(ctx); + sealToSubmit = new PendingSeal(lifecycleGeneration, detachedBucket, + detachedBucket.nextDeliveryTime(), + System.currentTimeMillis(), new CompletableFuture<>()); + pendingSeal = sealToSubmit; + pendingSealMemoryUsage.set(detachedBucket.getBufferMemoryUsage()); + pendingSealMessageCount.set(detachedBucket.size()); + stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.create); + } + + if (ledgerId >= lastMutableBucket.endLedgerId && !existBucket) { + lastMutableBucket.addMessage(ledgerId, entryId, deliverAt); + } else { + // Message index belongs to a previous bucket range, enter the shared queue directly. + // BucketDelayedMessageIndex remains the runtime source of truth while a seal is pending. + sharedBucketPriorityQueue.add(deliverAt, ledgerId, entryId); + } + index.track(ledgerId, entryId); log.debug() .attr("ledgerId", ledgerId) .attr("entryId", entryId) .attr("deliveryInMs", deliverAt - clock.millis()) .log("Add message"); - updateTimer(); + updateTimer(); + } + if (sealToSubmit != null) { + submitSeal(sealToSubmit); + } return true; } + private void submitSeal(PendingSeal seal) { + try { + snapshotBuildExecutor.execute(() -> buildAndCommitSeal(seal)); + } catch (RejectedExecutionException e) { + handleSealSubmissionFailure(seal, e); + } catch (RuntimeException e) { + handleSealSubmissionFailure(seal, e); + } + } + + private void handleSealSubmissionFailure(PendingSeal seal, RuntimeException error) { + boolean handled = false; + synchronized (this) { + if (pendingSeal == seal) { + if (lifecycle == Lifecycle.OPEN && lifecycleGeneration == seal.generation()) { + seal.bucket().moveAllMessagesToSharedQueue(sharedBucketPriorityQueue); + rescheduleTimer(0); + } + pendingSeal = null; + pendingSealMemoryUsage.set(0); + pendingSealMessageCount.set(0); + handled = true; + } + } + if (handled) { + log.warn() + .attr("startLedgerId", seal.bucket().startLedgerId) + .attr("endLedgerId", seal.bucket().endLedgerId) + .exception(error) + .log("Snapshot build executor rejected delayed-delivery bucket seal; using memory mode"); + stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.create); + finishSeal(seal); + } + } + + private void buildAndCommitSeal(PendingSeal seal) { + MutableBucket.SnapshotBuildResult result; + try { + result = seal.bucket().buildSnapshot(timeStepPerBucketSnapshotSegmentInMillis, + maxIndexesPerBucketSnapshotSegment); + if (result == null) { + throw new IllegalStateException("Detached mutable bucket is empty"); + } + } catch (MutableBucket.SnapshotBuildException e) { + handleSealBuildFailure(seal, e); + if (e.getCause() instanceof Error error) { + throw error; + } + return; + } catch (RuntimeException e) { + handleSealBuildFailure(seal, e); + return; + } + + CompletableFuture persistencePromise = null; + Throwable commitError = null; + synchronized (this) { + if (pendingSeal == seal && lifecycle == Lifecycle.OPEN + && lifecycleGeneration == seal.generation()) { + try { + result.addFirstSegmentTo(sharedBucketPriorityQueue); + persistencePromise = new CompletableFuture<>(); + result.bucket().setSnapshotCreateFuture(persistencePromise); + afterCreateImmutableBucket(Pair.of(result.bucket(), result.firstSegmentLastIndex()), + seal.startTime()); + pendingSeal = null; + pendingSealMemoryUsage.set(0); + pendingSealMessageCount.set(0); + triggerTrimAndMergeIfNeeded(); + rescheduleTimer(0); + } catch (RuntimeException e) { + commitError = e; + persistencePromise = null; + result.addAllSegmentsTo(sharedBucketPriorityQueue); + removeBucket(Range.closed(result.bucket().getStartLedgerId(), + result.bucket().getEndLedgerId())); + snapshotSegmentLastIndexMap.remove(new SnapshotKey( + result.firstSegmentLastIndex().getLedgerId(), + result.firstSegmentLastIndex().getEntryId())); + pendingSeal = null; + pendingSealMemoryUsage.set(0); + pendingSealMessageCount.set(0); + rescheduleTimer(0); + } + } else if (pendingSeal == seal) { + pendingSeal = null; + pendingSealMemoryUsage.set(0); + pendingSealMessageCount.set(0); + } + } + + if (persistencePromise != null) { + startSnapshotPersistence(result, persistencePromise); + } else if (commitError != null) { + log.error() + .attr("startLedgerId", seal.bucket().startLedgerId) + .attr("endLedgerId", seal.bucket().endLedgerId) + .exception(commitError) + .log("Failed to commit delayed-delivery bucket snapshot build; using memory mode"); + stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.create); + } + finishSeal(seal); + } + + private void startSnapshotPersistence(MutableBucket.SnapshotBuildResult result, + CompletableFuture persistencePromise) { + final CompletableFuture persistenceFuture; + try { + persistenceFuture = result.bucket().asyncSaveBucketSnapshot( + result.snapshotMetadata(), result.snapshotSegments()); + if (persistenceFuture == null) { + throw new NullPointerException("Snapshot storage returned a null future"); + } + } catch (Throwable t) { + persistencePromise.completeExceptionally(t); + rescheduleAfterAsyncOperation(); + return; + } + persistenceFuture.whenComplete((bucketId, error) -> { + if (error == null) { + persistencePromise.complete(bucketId); + } else { + persistencePromise.completeExceptionally(error); + } + rescheduleAfterAsyncOperation(); + }); + } + + private synchronized void rescheduleAfterAsyncOperation() { + if (lifecycle == Lifecycle.OPEN) { + rescheduleTimer(0); + } + } + + private void handleSealBuildFailure(PendingSeal seal, RuntimeException error) { + boolean handled = false; + synchronized (this) { + if (pendingSeal == seal) { + if (lifecycle == Lifecycle.OPEN && lifecycleGeneration == seal.generation()) { + if (error instanceof MutableBucket.SnapshotBuildException buildException) { + buildException.restoreTo(sharedBucketPriorityQueue); + } else { + seal.bucket().moveAllMessagesToSharedQueue(sharedBucketPriorityQueue); + } + rescheduleTimer(0); + } + pendingSeal = null; + pendingSealMemoryUsage.set(0); + pendingSealMessageCount.set(0); + handled = true; + } + } + if (handled) { + log.error() + .attr("startLedgerId", seal.bucket().startLedgerId) + .attr("endLedgerId", seal.bucket().endLedgerId) + .exception(error) + .log("Failed to build delayed-delivery bucket snapshot; using memory mode"); + stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.create); + finishSeal(seal); + } + } + + private void finishSeal(PendingSeal seal) { + try { + seal.bucket().close(); + } catch (RuntimeException error) { + log.warn().exception(error).log("Failed to release detached delayed-delivery bucket"); + } finally { + seal.completion().complete(null); + } + } + + private void triggerTrimAndMergeIfNeeded() { + if (maxNumBuckets > 0 && bucketsCount.get() > maxNumBuckets + && (trimFuture == null || trimFuture.isDone())) { + trimFuture = asyncTrimImmutableBuckets() + .thenCompose(ignore -> asyncMergeBucketSnapshot()) + .whenComplete((ignore, t) -> { + if (t != null) { + log.warn().exception(t).log("Failed to trim or merge bucket snapshots"); + } + }); + } + } + @VisibleForTesting synchronized List selectMergedBuckets(final List values, int mergeNum) { if (values.size() < 2 || mergeNum < 2) { @@ -622,6 +884,9 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot(List 0 && nextDeliveryTime() <= cutoffTime; @@ -633,19 +898,16 @@ public synchronized boolean hasMessageAvailable() { @Override protected synchronized long nextDeliveryTime() { - if (lastMutableBucket.isEmpty() && !sharedBucketPriorityQueue.isEmpty()) { - return sharedBucketPriorityQueue.peekN1(); - } else if (sharedBucketPriorityQueue.isEmpty() && !lastMutableBucket.isEmpty()) { - return lastMutableBucket.nextDeliveryTime(); - } else if (lastMutableBucket.isEmpty() && sharedBucketPriorityQueue.isEmpty()) { - // numberDelayedMessages can be > 0 while both queues are empty (e.g. remaining - // messages live in not-yet-loaded snapshot segments). Returning Long.MAX_VALUE - // signals "no imminent delivery" without throwing on the empty queues. - return Long.MAX_VALUE; + long timestamp = lastMutableBucket.isEmpty() ? Long.MAX_VALUE : lastMutableBucket.nextDeliveryTime(); + if (!sharedBucketPriorityQueue.isEmpty()) { + timestamp = Math.min(timestamp, sharedBucketPriorityQueue.peekN1()); + } + if (pendingSeal != null) { + timestamp = Math.min(timestamp, pendingSeal.nextDeliveryTime()); } - long timestamp = lastMutableBucket.nextDeliveryTime(); - long bucketTimestamp = sharedBucketPriorityQueue.peekN1(); - return Math.min(timestamp, bucketTimestamp); + // numberDelayedMessages can be > 0 while all in-memory queues are empty because later + // messages can live in not-yet-loaded snapshot segments. + return timestamp; } @Override @@ -655,11 +917,15 @@ public long getNumberOfDelayedMessages() { @Override public long getBufferMemoryUsage() { - return this.lastMutableBucket.getBufferMemoryUsage() + sharedBucketPriorityQueue.bytesCapacity(); + return this.lastMutableBucket.getBufferMemoryUsage() + sharedBucketPriorityQueue.bytesCapacity() + + pendingSealMemoryUsage.get(); } @Override public synchronized NavigableSet getScheduledMessages(int maxMessages) { + if (lifecycle != Lifecycle.OPEN) { + return Collections.emptyNavigableSet(); + } if (!checkPendingLoadDone()) { log.debug("Skip getScheduledMessages to wait for bucket snapshot load finish"); return Collections.emptyNavigableSet(); @@ -756,7 +1022,7 @@ public synchronized NavigableSet getScheduledMessages(int maxMessages) stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.load, System.currentTimeMillis() - loadStartTime); } - rescheduleTimer(0); + rescheduleAfterAsyncOperation(); }); if (!checkPendingLoadDone() || loadFuture.isCompletedExceptionally()) { @@ -792,27 +1058,69 @@ public boolean shouldPauseAllDeliveries() { } @Override - public synchronized CompletableFuture clear() { - // Wait for any in-flight trim+merge to settle, then clear. - // Reuse trimFuture to block new triggers until the clear chain completes. - CompletableFuture before = trimFuture != null && !trimFuture.isDone() - ? trimFuture : CompletableFuture.completedFuture(null); - trimFuture = before - .exceptionally(t -> { - log.warn().exception(t).log("Trim/merge buckets failed, but still clear"); - return null; - }) - .thenCompose(__ -> { - synchronized (BucketDelayedDeliveryTracker.this) { - CompletableFuture future = cleanImmutableBuckets(); - sharedBucketPriorityQueue.clear(); - index.clear(); - lastMutableBucket.clear(); - snapshotSegmentLastIndexMap.clear(); - return future; - } - }); - return trimFuture; + public CompletableFuture clear() { + final CompletableFuture result; + final CompletableFuture before; + final long clearGeneration; + synchronized (this) { + if (lifecycle == Lifecycle.CLOSING || lifecycle == Lifecycle.CLOSED) { + return closeFuture != null ? closeFuture : CompletableFuture.completedFuture(null); + } + if (lifecycle == Lifecycle.CLEARING) { + return clearFuture; + } + + lifecycle = Lifecycle.CLEARING; + clearGeneration = ++lifecycleGeneration; + // Reset the current timeout before temporarily rejecting timer callbacks. New messages + // can arm the tracker again after this clear operation reopens it. + super.close(); + result = new CompletableFuture<>(); + clearFuture = result; + + List> operations = new ArrayList<>(); + if (pendingSeal != null) { + operations.add(pendingSeal.completion()); + } + if (pendingLoad != null) { + operations.add(pendingLoad); + } + if (trimFuture != null && !trimFuture.isDone()) { + operations.add(trimFuture); + } + before = FutureUtil.waitForAll(operations).exceptionally(t -> { + log.warn().exception(t).log("An asynchronous bucket operation failed, but still clear"); + return null; + }); + // Block new trim/merge triggers until this clear operation has settled. + trimFuture = result; + } + + before.thenCompose(__ -> { + synchronized (BucketDelayedDeliveryTracker.this) { + if (lifecycle != Lifecycle.CLEARING || lifecycleGeneration != clearGeneration) { + return CompletableFuture.completedFuture(null); + } + CompletableFuture deleteFuture = cleanImmutableBuckets(); + sharedBucketPriorityQueue.clear(); + index.clear(); + lastMutableBucket.clear(); + snapshotSegmentLastIndexMap.clear(); + return deleteFuture; + } + }).whenComplete((__, error) -> { + synchronized (BucketDelayedDeliveryTracker.this) { + if (lifecycle == Lifecycle.CLEARING && lifecycleGeneration == clearGeneration) { + lifecycle = Lifecycle.OPEN; + } + } + if (error == null) { + result.complete(null); + } else { + result.completeExceptionally(error); + } + }); + return result; } @Override @@ -823,19 +1131,78 @@ public void close() { @Override public CompletableFuture closeAsync() { - List> completableFutures; + final CompletableFuture result; + final CompletableFuture before; synchronized (this) { + if (closeFuture != null) { + return closeFuture; + } + + lifecycle = Lifecycle.CLOSING; + lifecycleGeneration++; super.close(); - lastMutableBucket.close(); - sharedBucketPriorityQueue.close(); - completableFutures = immutableBuckets.asMapOfRanges().values().stream() - .map(bucket -> bucket.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE)).toList(); - } - return FutureUtil.waitForAll(completableFutures) - .exceptionally(e -> { - log.warn().exception(e).log("Failed wait to snapshot generate"); - return null; - }); + result = new CompletableFuture<>(); + closeFuture = result; + + List> operations = new ArrayList<>(); + if (pendingSeal != null) { + operations.add(pendingSeal.completion()); + } + if (pendingLoad != null) { + operations.add(pendingLoad); + } + if (trimFuture != null && !trimFuture.isDone()) { + operations.add(trimFuture); + } + if (clearFuture != null && !clearFuture.isDone() && clearFuture != trimFuture) { + operations.add(clearFuture); + } + before = FutureUtil.waitForAll(operations).exceptionally(t -> { + log.warn().exception(t).log("Failed waiting for delayed-delivery bucket operations during close"); + return null; + }); + } + + before.thenCompose(__ -> { + List> snapshotFutures; + synchronized (BucketDelayedDeliveryTracker.this) { + snapshotFutures = immutableBuckets.asMapOfRanges().values().stream() + .map(bucket -> bucket.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE)).toList(); + } + return FutureUtil.waitForAll(snapshotFutures).exceptionally(error -> { + log.warn().exception(error).log("Failed waiting for bucket snapshot creation during close"); + return null; + }); + }).whenComplete((__, error) -> { + Throwable completionError = error; + synchronized (BucketDelayedDeliveryTracker.this) { + try { + lastMutableBucket.close(); + } catch (Throwable closeError) { + if (completionError == null) { + completionError = closeError; + } else { + completionError.addSuppressed(closeError); + } + } + try { + sharedBucketPriorityQueue.close(); + } catch (Throwable closeError) { + if (completionError == null) { + completionError = closeError; + } else { + completionError.addSuppressed(closeError); + } + } + lifecycle = Lifecycle.CLOSED; + } + if (completionError == null) { + result.complete(null); + } else { + result.completeExceptionally(completionError); + } + }); + return result; } private CompletableFuture cleanImmutableBuckets() { @@ -861,7 +1228,8 @@ public synchronized boolean containsMessage(long ledgerId, long entryId) { public Map genTopicMetricMap() { stats.recordNumOfBuckets((int) (bucketsCount.get() + 1)); - stats.recordDelayedMessageIndexLoaded(this.sharedBucketPriorityQueue.size() + this.lastMutableBucket.size()); + stats.recordDelayedMessageIndexLoaded(this.sharedBucketPriorityQueue.size() + this.lastMutableBucket.size() + + pendingSealMessageCount.get()); stats.recordBucketSnapshotSizeBytes(totalSnapshotLengthBytes.get()); return stats.genTopicMetricMap(); } 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 index 925c4756019c7..4f27af5bdf6da 100644 --- 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 @@ -20,7 +20,6 @@ import static org.apache.bookkeeper.mledger.util.Futures.executeWithRetry; import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.DELAYED_BUCKET_KEY_PREFIX; -import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.NULL_LONG_PROMISE; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; @@ -349,8 +348,22 @@ CompletableFuture asyncDeleteBucketSnapshot(BucketDelayedMessageIndexStats CompletableFuture clear(BucketDelayedMessageIndexStats stats) { delayedIndexBitMap.clear(); - return getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE).exceptionally(e -> null) - .thenCompose(__ -> asyncDeleteBucketSnapshot(stats)); + Optional> createFuture = getSnapshotCreateFuture(); + if (createFuture.isEmpty()) { + // Recovered buckets don't have a create future; their id is read from the cursor property. + return asyncDeleteBucketSnapshot(stats); + } + return createFuture.get() + .handle((createdBucketId, error) -> getBucketId().orElse(error == null ? createdBucketId : null)) + .thenCompose(createdBucketId -> { + if (createdBucketId == null || createdBucketId < 0) { + return CompletableFuture.completedFuture(null); + } + if (getBucketId().isEmpty()) { + setBucketId(createdBucketId); + } + return asyncDeleteBucketSnapshot(stats); + }); } protected CompletableFuture asyncUpdateSnapshotLength() { 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 index 0d3630987af8c..7fa30370f6066 100644 --- 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 @@ -50,27 +50,41 @@ class MutableBucket implements AutoCloseable { this.priorityQueue = new TripleLongPriorityQueue(); } - Pair sealBucketAndAsyncPersistent( - long timeStepPerBucketSnapshotSegment, - int maxIndexesPerBucketSnapshotSegment, - TripleLongPriorityQueue sharedQueue) { - return createImmutableBucketAndAsyncPersistent(timeStepPerBucketSnapshotSegment, - maxIndexesPerBucketSnapshotSegment, sharedQueue, - TripleLongPriorityDelayedIndexQueue.wrap(priorityQueue), startLedgerId, endLedgerId); - } - Pair createImmutableBucketAndAsyncPersistent( final long timeStepPerBucketSnapshotSegment, final int maxIndexesPerBucketSnapshotSegment, TripleLongPriorityQueue sharedQueue, DelayedIndexQueue delayedIndexQueue, final long startLedgerId, final long endLedgerId) { - log.debug() - .attr("dispatcher", ctx.dispatcherName()) - .attr("startLedgerId", startLedgerId) - .attr("endLedgerId", endLedgerId) - .log("Creating bucket snapshot"); - if (delayedIndexQueue.isEmpty()) { + SnapshotBuildResult result = buildSnapshot(timeStepPerBucketSnapshotSegment, + maxIndexesPerBucketSnapshotSegment, delayedIndexQueue, startLedgerId, endLedgerId); + if (result == null) { return null; } + + result.addFirstSegmentTo(sharedQueue); + CompletableFuture future = result.bucket().asyncSaveBucketSnapshot( + result.snapshotMetadata(), result.snapshotSegments()); + result.bucket().setSnapshotCreateFuture(future); + return Pair.of(result.bucket(), result.firstSegmentLastIndex()); + } + + SnapshotBuildResult buildSnapshot(long timeStepPerBucketSnapshotSegment, + int maxIndexesPerBucketSnapshotSegment) { + return buildSnapshot(timeStepPerBucketSnapshotSegment, maxIndexesPerBucketSnapshotSegment, + TripleLongPriorityDelayedIndexQueue.wrap(priorityQueue), startLedgerId, endLedgerId); + } + + private SnapshotBuildResult buildSnapshot( + final long timeStepPerBucketSnapshotSegment, final int maxIndexesPerBucketSnapshotSegment, + DelayedIndexQueue delayedIndexQueue, final long startLedgerId, final long endLedgerId) { + log.debug() + .attr("dispatcher", ctx.dispatcherName()) + .attr("startLedgerId", startLedgerId) + .attr("endLedgerId", endLedgerId) + .log("Building bucket snapshot"); + if (delayedIndexQueue.isEmpty()) { + return null; + } + long numMessages = 0; List bucketSnapshotSegments = new ArrayList<>(); @@ -84,91 +98,86 @@ Pair createImmutableBucketAndAsyncPersistent( List firstScheduleTimestamps = new ArrayList<>(); long currentTimestampUpperLimit = 0; long currentFirstTimestamp = 0L; - while (!delayedIndexQueue.isEmpty()) { - final long timestamp = delayedIndexQueue.peekTimestamp(); - if (currentTimestampUpperLimit == 0) { - currentFirstTimestamp = timestamp; - firstScheduleTimestamps.add(currentFirstTimestamp); - currentTimestampUpperLimit = timestamp + timeStepPerBucketSnapshotSegment - 1; - } - - DelayedIndex delayedIndex = snapshotSegment.addIndexe(); - delayedIndexQueue.popToObject(delayedIndex); - - final long ledgerId = delayedIndex.getLedgerId(); - final long entryId = delayedIndex.getEntryId(); - - checkArgument(ledgerId >= startLedgerId && ledgerId <= endLedgerId); - - // Move first segment of bucket snapshot to sharedBucketPriorityQueue - if (segmentMetadataList.size() == 0) { - sharedQueue.add(timestamp, ledgerId, entryId); - } - - bitMap.computeIfAbsent(ledgerId, k -> LongBitmaps.create()).add(entryId); - - numMessages++; - - if (delayedIndexQueue.isEmpty() || delayedIndexQueue.peekTimestamp() > currentTimestampUpperLimit - || (maxIndexesPerBucketSnapshotSegment != -1 - && snapshotSegment.getIndexesCount() >= maxIndexesPerBucketSnapshotSegment)) { - segmentMetadata.setMaxScheduleTimestamp(timestamp); - segmentMetadata.setMinScheduleTimestamp(currentFirstTimestamp); - currentTimestampUpperLimit = 0; - - Iterator> iterator = bitMap.entrySet().iterator(); - while (iterator.hasNext()) { - final var entry = iterator.next(); - final var lId = entry.getKey(); - final var bm = entry.getValue(); - segmentMetadata.putDelayedIndexBitMap(lId, bm.serialize()); - immutableBucketBitMap.compute(lId, (__, bm0) -> { - if (bm0 == null) { - return bm; - } - bm0.or(bm); - return bm0; - }); - iterator.remove(); + try { + while (!delayedIndexQueue.isEmpty()) { + final long timestamp = delayedIndexQueue.peekTimestamp(); + if (currentTimestampUpperLimit == 0) { + currentFirstTimestamp = timestamp; + firstScheduleTimestamps.add(currentFirstTimestamp); + currentTimestampUpperLimit = timestamp + timeStepPerBucketSnapshotSegment - 1; } - segmentMetadataList.add(segmentMetadata); - segmentMetadata = new SnapshotSegmentMetadata(); - - bucketSnapshotSegments.add(snapshotSegment); - snapshotSegment = new SnapshotSegment(); + DelayedIndex delayedIndex = snapshotSegment.addIndexe(); + delayedIndexQueue.popToObject(delayedIndex); + + final long ledgerId = delayedIndex.getLedgerId(); + final long entryId = delayedIndex.getEntryId(); + + checkArgument(ledgerId >= startLedgerId && ledgerId <= endLedgerId); + + bitMap.computeIfAbsent(ledgerId, k -> LongBitmaps.create()).add(entryId); + + numMessages++; + + if (delayedIndexQueue.isEmpty() || delayedIndexQueue.peekTimestamp() > currentTimestampUpperLimit + || (maxIndexesPerBucketSnapshotSegment != -1 + && snapshotSegment.getIndexesCount() >= maxIndexesPerBucketSnapshotSegment)) { + segmentMetadata.setMaxScheduleTimestamp(timestamp); + segmentMetadata.setMinScheduleTimestamp(currentFirstTimestamp); + currentTimestampUpperLimit = 0; + + Iterator> iterator = bitMap.entrySet().iterator(); + while (iterator.hasNext()) { + final var entry = iterator.next(); + final var lId = entry.getKey(); + final var bm = entry.getValue(); + segmentMetadata.putDelayedIndexBitMap(lId, bm.serialize()); + immutableBucketBitMap.compute(lId, (__, bm0) -> { + if (bm0 == null) { + return bm; + } + bm0.or(bm); + return bm0; + }); + iterator.remove(); + } + + segmentMetadataList.add(segmentMetadata); + segmentMetadata = new SnapshotSegmentMetadata(); + + bucketSnapshotSegments.add(snapshotSegment); + snapshotSegment = new SnapshotSegment(); + } } - } - SnapshotMetadata bucketSnapshotMetadata = new SnapshotMetadata(); - for (SnapshotSegmentMetadata sm : segmentMetadataList) { - bucketSnapshotMetadata.addMetadata().copyFrom(sm); - } - - final int lastSegmentEntryId = segmentMetadataList.size(); + SnapshotMetadata bucketSnapshotMetadata = new SnapshotMetadata(); + for (SnapshotSegmentMetadata sm : segmentMetadataList) { + bucketSnapshotMetadata.addMetadata().copyFrom(sm); + } - ImmutableBucket bucket = new ImmutableBucket(ctx, startLedgerId, endLedgerId); - bucket.setCurrentSegmentEntryId(1); - bucket.setNumberBucketDelayedMessages(numMessages); - bucket.setLastSegmentEntryId(lastSegmentEntryId); - bucket.setFirstScheduleTimestamps(firstScheduleTimestamps); - bucket.setDelayedIndexBitMap(immutableBucketBitMap); + final int lastSegmentEntryId = segmentMetadataList.size(); - // Skip first segment, because it has already been loaded - List snapshotSegments = bucketSnapshotSegments.subList(1, bucketSnapshotSegments.size()); - bucket.setSnapshotSegments(snapshotSegments); + ImmutableBucket bucket = new ImmutableBucket(ctx, startLedgerId, endLedgerId); + bucket.setCurrentSegmentEntryId(1); + bucket.setNumberBucketDelayedMessages(numMessages); + bucket.setLastSegmentEntryId(lastSegmentEntryId); + bucket.setFirstScheduleTimestamps(firstScheduleTimestamps); + bucket.setDelayedIndexBitMap(immutableBucketBitMap); - // Add the first snapshot segment last message to snapshotSegmentLastMessageTable - checkArgument(!bucketSnapshotSegments.isEmpty()); - SnapshotSegment firstSnapshotSegment = bucketSnapshotSegments.get(0); - DelayedIndex lastDelayedIndex = firstSnapshotSegment.getIndexeAt(firstSnapshotSegment.getIndexesCount() - 1); - Pair result = Pair.of(bucket, lastDelayedIndex); + // Skip the first segment because the tracker loads it into the shared queue before committing the bucket. + bucket.setSnapshotSegments(new ArrayList<>( + bucketSnapshotSegments.subList(1, bucketSnapshotSegments.size()))); - CompletableFuture future = bucket.asyncSaveBucketSnapshot( - bucketSnapshotMetadata, bucketSnapshotSegments); - bucket.setSnapshotCreateFuture(future); + checkArgument(!bucketSnapshotSegments.isEmpty()); + SnapshotSegment firstSnapshotSegment = bucketSnapshotSegments.get(0); + DelayedIndex lastDelayedIndex = + firstSnapshotSegment.getIndexeAt(firstSnapshotSegment.getIndexesCount() - 1); - return result; + return new SnapshotBuildResult(bucket, bucketSnapshotMetadata, + bucketSnapshotSegments, lastDelayedIndex); + } catch (Throwable t) { + throw new SnapshotBuildException(t, bucketSnapshotSegments, snapshotSegment, delayedIndexQueue); + } } void moveScheduledMessageToSharedQueue(long cutoffTime, TripleLongPriorityQueue sharedBucketPriorityQueue) { @@ -186,6 +195,13 @@ void moveScheduledMessageToSharedQueue(long cutoffTime, TripleLongPriorityQueue } } + void moveAllMessagesToSharedQueue(TripleLongPriorityQueue sharedBucketPriorityQueue) { + while (!priorityQueue.isEmpty()) { + sharedBucketPriorityQueue.add(priorityQueue.peekN1(), priorityQueue.peekN2(), priorityQueue.peekN3()); + priorityQueue.pop(); + } + } + void resetLastMutableBucketRange() { this.startLedgerId = -1L; this.endLedgerId = -1L; @@ -223,4 +239,58 @@ void addMessage(long ledgerId, long entryId, long deliverAt) { } this.endLedgerId = ledgerId; } + + static record SnapshotBuildResult(ImmutableBucket bucket, + SnapshotMetadata snapshotMetadata, + List snapshotSegments, + DelayedIndex firstSegmentLastIndex) { + + void addFirstSegmentTo(TripleLongPriorityQueue sharedQueue) { + for (DelayedIndex delayedIndex : snapshotSegments.get(0).getIndexesList()) { + sharedQueue.add(delayedIndex.getTimestamp(), delayedIndex.getLedgerId(), delayedIndex.getEntryId()); + } + } + + void addAllSegmentsTo(TripleLongPriorityQueue sharedQueue) { + for (SnapshotSegment segment : snapshotSegments) { + for (DelayedIndex delayedIndex : segment.getIndexesList()) { + sharedQueue.add(delayedIndex.getTimestamp(), delayedIndex.getLedgerId(), delayedIndex.getEntryId()); + } + } + } + } + + static final class SnapshotBuildException extends RuntimeException { + private final List completedSegments; + private final SnapshotSegment currentSegment; + private final DelayedIndexQueue remainingIndexes; + + SnapshotBuildException(Throwable cause, List completedSegments, + SnapshotSegment currentSegment, DelayedIndexQueue remainingIndexes) { + super(cause); + this.completedSegments = completedSegments; + this.currentSegment = currentSegment; + this.remainingIndexes = remainingIndexes; + } + + void restoreTo(TripleLongPriorityQueue sharedQueue) { + for (SnapshotSegment segment : completedSegments) { + restoreSegment(segment, sharedQueue); + } + if (completedSegments.isEmpty() + || completedSegments.get(completedSegments.size() - 1) != currentSegment) { + restoreSegment(currentSegment, sharedQueue); + } + while (!remainingIndexes.isEmpty()) { + DelayedIndex delayedIndex = remainingIndexes.pop(); + sharedQueue.add(delayedIndex.getTimestamp(), delayedIndex.getLedgerId(), delayedIndex.getEntryId()); + } + } + + private static void restoreSegment(SnapshotSegment segment, TripleLongPriorityQueue sharedQueue) { + for (DelayedIndex delayedIndex : segment.getIndexesList()) { + sharedQueue.add(delayedIndex.getTimestamp(), delayedIndex.getLedgerId(), delayedIndex.getEntryId()); + } + } + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java index 54140cb2bfdf7..396b70e166d08 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java @@ -22,7 +22,9 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotSame; @@ -35,6 +37,7 @@ import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.time.Clock; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -46,9 +49,12 @@ import java.util.TreeSet; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import lombok.Cleanup; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.bookkeeper.mledger.ManagedLedger; @@ -1286,6 +1292,264 @@ public void testLateSnapshotLengthUpdateAfterTrimDoesNotInflateCounter() throws } } + @Test + public void testSnapshotBuildIsSubmittedOutsideTrackerMonitor() throws Exception { + AtomicReference buildTask = new AtomicReference<>(); + AtomicReference trackerReference = new AtomicReference<>(); + Executor executor = command -> { + Assert.assertFalse(Thread.holdsLock(trackerReference.get()), + "Snapshot build was submitted while holding the tracker monitor"); + Assert.assertTrue(buildTask.compareAndSet(null, command)); + }; + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(executor, now, storage); + trackerReference.set(tracker); + when(storage.createBucketSnapshot(any(), any(), any(), any(), any())).thenAnswer(__ -> { + Assert.assertFalse(Thread.holdsLock(tracker), + "Snapshot serialization was invoked while holding the tracker monitor"); + return CompletableFuture.completedFuture(1L); + }); + + addMessagesAndTriggerSeal(tracker); + + assertTrue(buildTask.get() != null, "The detached bucket should have been submitted for background build"); + assertEquals(tracker.getLastMutableBucket().size(), 1); + assertEquals(tracker.getNumberOfDelayedMessages(), 4); + assertTrue(tracker.getImmutableBuckets().asMapOfRanges().isEmpty()); + + buildTask.get().run(); + + assertEquals(tracker.getImmutableBuckets().asMapOfRanges().size(), 1); + tracker.close(); + } + + @Test + public void testRejectedSealSubmissionFallsBackToMemoryMode() throws Exception { + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + Executor rejectingExecutor = command -> { + throw new RejectedExecutionException("test rejection"); + }; + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(rejectingExecutor, now, storage); + + addMessagesAndTriggerSeal(tracker); + + assertTrue(tracker.getImmutableBuckets().asMapOfRanges().isEmpty()); + assertEquals(tracker.getSharedBucketPriorityQueue().size(), 3); + assertEquals(tracker.getLastMutableBucket().size(), 1); + assertEquals(tracker.getNumberOfDelayedMessages(), 4); + + now.set(2_000); + assertEquals(tracker.getScheduledMessages(10).size(), 4); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + tracker.close(); + } + + @Test + public void testSnapshotBuildFailureRestoresAllIndexes() throws Exception { + AtomicReference buildTask = new AtomicReference<>(); + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(buildTask::set, now, storage); + + for (int entryId = 0; entryId < 3; entryId++) { + assertTrue(tracker.addMessage(1, entryId, 1_000)); + } + // Corrupt the test bucket range so the builder fails after destructively popping an index. + tracker.getLastMutableBucket().endLedgerId = 0; + assertTrue(tracker.addMessage(2, 0, 1_000)); + buildTask.get().run(); + + assertTrue(tracker.getImmutableBuckets().asMapOfRanges().isEmpty()); + assertEquals(tracker.getSharedBucketPriorityQueue().size(), 3); + assertEquals(tracker.getLastMutableBucket().size(), 1); + assertEquals(tracker.getNumberOfDelayedMessages(), 4); + + now.set(2_000); + assertEquals(tracker.getScheduledMessages(10).size(), 4); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + tracker.close(); + } + + @Test + public void testClearDuringSealDoesNotResurrectBucket() throws Exception { + AtomicReference buildTask = new AtomicReference<>(); + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(buildTask::set, now, storage); + + addMessagesAndTriggerSeal(tracker); + CompletableFuture clearFuture = tracker.clear(); + assertFalse("Clear should wait for the detached bucket build to observe its generation fence", + clearFuture.isDone()); + + buildTask.get().run(); + clearFuture.get(5, TimeUnit.SECONDS); + + assertTrue(tracker.getImmutableBuckets().asMapOfRanges().isEmpty()); + assertEquals(tracker.getSharedBucketPriorityQueue().size(), 0); + assertEquals(tracker.getLastMutableBucket().size(), 0); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + verify(storage, never()).createBucketSnapshot(any(), any(), any(), any(), any()); + + assertTrue(tracker.addMessage(3, 0, 3_000), "Tracker should be reusable after clear"); + tracker.close(); + } + + @Test + public void testPersistenceFailureDuringClearDoesNotRestoreMessages() throws Exception { + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + CompletableFuture persistenceFuture = new CompletableFuture<>(); + when(storage.createBucketSnapshot(any(), any(), any(), any(), any())).thenReturn(persistenceFuture); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(Runnable::run, now, storage); + + addMessagesAndTriggerSeal(tracker); + assertEquals(tracker.getImmutableBuckets().asMapOfRanges().size(), 1); + + CompletableFuture clearFuture = tracker.clear(); + assertFalse("Clear should wait for the in-flight snapshot persistence", clearFuture.isDone()); + persistenceFuture.completeExceptionally(new RuntimeException("test persistence failure")); + clearFuture.get(5, TimeUnit.SECONDS); + + assertTrue(tracker.getImmutableBuckets().asMapOfRanges().isEmpty()); + assertEquals(tracker.getSharedBucketPriorityQueue().size(), 0); + assertEquals(tracker.getLastMutableBucket().size(), 0); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + verify(storage, never()).deleteBucketSnapshot(anyLong()); + tracker.close(); + } + + @Test + public void testClearDeletesSnapshotWhenCreationChainFailsAfterAllocatingId() throws Exception { + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + CompletableFuture persistenceFuture = new CompletableFuture<>(); + when(storage.createBucketSnapshot(any(), any(), any(), any(), any())).thenReturn(persistenceFuture); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(Runnable::run, now, storage); + + addMessagesAndTriggerSeal(tracker); + ImmutableBucket bucket = tracker.getImmutableBuckets().asMapOfRanges().values().iterator().next(); + bucket.setBucketId(42L); + + CompletableFuture clearFuture = tracker.clear(); + persistenceFuture.completeExceptionally(new RuntimeException("test creation-chain failure")); + clearFuture.get(5, TimeUnit.SECONDS); + + verify(storage).deleteBucketSnapshot(42L); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + tracker.close(); + } + + @Test + public void testCloseAsyncDuringSealDoesNotBlockCaller() throws Exception { + AtomicReference buildTask = new AtomicReference<>(); + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(buildTask::set, now, storage); + + addMessagesAndTriggerSeal(tracker); + CompletableFuture closeFuture = tracker.closeAsync(); + assertFalse("closeAsync should return without waiting on the caller thread", closeFuture.isDone()); + + buildTask.get().run(); + closeFuture.get(5, TimeUnit.SECONDS); + + verify(storage, never()).createBucketSnapshot(any(), any(), any(), any(), any()); + assertFalse("Closed tracker must reject new delayed messages", tracker.addMessage(3, 0, 3_000)); + Assert.assertSame(tracker.closeAsync(), closeFuture); + } + + @Test + public void testDueMessagesDuringSealRemainReachableAndDeduplicated() throws Exception { + AtomicReference buildTask = new AtomicReference<>(); + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(buildTask::set, now, storage); + + addMessagesAndTriggerSeal(tracker); + assertTrue(tracker.addMessage(1, 3, 1_000)); + assertTrue(tracker.addMessage(1, 3, 1_000)); + assertEquals(tracker.getNumberOfDelayedMessages(), 5); + + now.set(2_000); + NavigableSet beforeBuild = tracker.getScheduledMessages(10); + assertEquals(beforeBuild.size(), 2); + + buildTask.get().run(); + NavigableSet afterBuild = tracker.getScheduledMessages(10); + assertEquals(afterBuild.size(), 3); + TreeSet delivered = new TreeSet<>(beforeBuild); + delivered.addAll(afterBuild); + assertEquals(delivered.size(), 5); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + tracker.close(); + } + + @Test + public void testLateMessageInSealingRangeSurvivesNextSealAndRemainsDeduplicated() throws Exception { + ArrayDeque buildTasks = new ArrayDeque<>(); + AtomicLong now = new AtomicLong(); + BucketSnapshotStorage storage = newCompletedSnapshotStorage(); + BucketDelayedDeliveryTracker tracker = newAsyncSealTracker(buildTasks::addLast, now, storage); + + addMessagesAndTriggerSeal(tracker); + assertTrue(tracker.addMessage(1, 3, 1_000)); + assertTrue(tracker.addMessage(1, 3, 1_000)); + assertTrue(tracker.addMessage(2, 1, 1_000)); + assertTrue(tracker.addMessage(2, 2, 1_000)); + assertEquals(tracker.getNumberOfDelayedMessages(), 7); + + buildTasks.removeFirst().run(); + assertTrue(tracker.addMessage(3, 0, 1_000)); + assertEquals(buildTasks.size(), 1, "The second mutable bucket should seal independently"); + buildTasks.removeFirst().run(); + + now.set(2_000); + NavigableSet delivered = tracker.getScheduledMessages(20); + assertEquals(delivered.size(), 8); + assertTrue(delivered.contains(PositionFactory.create(1, 3))); + assertEquals(tracker.getScheduledMessages(20).size(), 0); + assertEquals(tracker.getNumberOfDelayedMessages(), 0); + tracker.close(); + } + + private BucketDelayedDeliveryTracker newAsyncSealTracker(Executor executor, AtomicLong now, + BucketSnapshotStorage storage) throws Exception { + AbstractPersistentDispatcherMultipleConsumers testDispatcher = + mock(AbstractPersistentDispatcherMultipleConsumers.class); + ManagedCursor cursor = mock(ManagedCursor.class); + when(cursor.getName()).thenReturn("async-seal-cursor"); + when(cursor.getCursorProperties()).thenReturn(Map.of()); + when(cursor.putCursorProperty(any(), any())).thenReturn(CompletableFuture.completedFuture(null)); + when(cursor.removeCursorProperty(any())).thenReturn(CompletableFuture.completedFuture(null)); + doReturn(cursor).when(testDispatcher).getCursor(); + doReturn("persistent://public/default/async-seal / " + cursor.getName()) + .when(testDispatcher).getName(); + Clock testClock = mock(Clock.class); + when(testClock.millis()).thenAnswer(__ -> now.get()); + bucketSnapshotStorage = storage; + return new BucketDelayedDeliveryTracker(testDispatcher, mock(Timer.class), 1, testClock, + true, storage, 3, TimeUnit.HOURS.toMillis(1), -1, 50, executor); + } + + private static BucketSnapshotStorage newCompletedSnapshotStorage() { + BucketSnapshotStorage storage = mock(BucketSnapshotStorage.class); + when(storage.createBucketSnapshot(any(), any(), any(), any(), any())) + .thenReturn(CompletableFuture.completedFuture(1L)); + when(storage.getBucketSnapshotLength(anyLong())).thenReturn(CompletableFuture.completedFuture(0L)); + when(storage.deleteBucketSnapshot(anyLong())).thenReturn(CompletableFuture.completedFuture(null)); + return storage; + } + + private static void addMessagesAndTriggerSeal(BucketDelayedDeliveryTracker tracker) { + for (int entryId = 0; entryId < 3; entryId++) { + assertTrue(tracker.addMessage(1, entryId, 1_000)); + } + assertTrue(tracker.addMessage(2, 0, 1_000)); + } + private static void assertCountersConsistent(BucketDelayedDeliveryTracker tracker) { int liveBucketCount = tracker.getImmutableBuckets().asMapOfRanges().size(); long liveSnapshotLength = tracker.getImmutableBuckets().asMapOfRanges().values().stream()