diff --git a/.github/workflows/ci-go-functions.yaml b/.github/workflows/ci-go-functions.yaml index f4d1ae0b99887..a34cb1a90908e 100644 --- a/.github/workflows/ci-go-functions.yaml +++ b/.github/workflows/ci-go-functions.yaml @@ -32,7 +32,7 @@ concurrency: cancel-in-progress: true env: - MAVEN_OPTS: -Xss1500k -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 + MAVEN_OPTS: -Xss1500k -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 jobs: preconditions: diff --git a/.github/workflows/ci-maven-cache-update.yaml b/.github/workflows/ci-maven-cache-update.yaml index 87570586fde6e..a5a88e2e0b8a6 100644 --- a/.github/workflows/ci-maven-cache-update.yaml +++ b/.github/workflows/ci-maven-cache-update.yaml @@ -42,7 +42,7 @@ on: - cron: '30 */12 * * *' env: - MAVEN_OPTS: -Xss1500k -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 + MAVEN_OPTS: -Xss1500k -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 jobs: update-maven-dependencies-cache: diff --git a/.github/workflows/ci-owasp-dependency-check.yaml b/.github/workflows/ci-owasp-dependency-check.yaml index 194d88c582d42..dba1c92ca28dd 100644 --- a/.github/workflows/ci-owasp-dependency-check.yaml +++ b/.github/workflows/ci-owasp-dependency-check.yaml @@ -24,7 +24,7 @@ on: workflow_dispatch: env: - MAVEN_OPTS: -Xss1500k -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 + MAVEN_OPTS: -Xss1500k -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 jobs: run-owasp-dependency-check: diff --git a/.github/workflows/pulsar-ci-flaky.yaml b/.github/workflows/pulsar-ci-flaky.yaml index acfa66ff43c74..ea19bc3307819 100644 --- a/.github/workflows/pulsar-ci-flaky.yaml +++ b/.github/workflows/pulsar-ci-flaky.yaml @@ -36,7 +36,7 @@ concurrency: cancel-in-progress: true env: - MAVEN_OPTS: -Xss1500k -Xmx1024m -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 + MAVEN_OPTS: -Xss1500k -Xmx1024m -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 # defines the retention period for the intermediate build artifacts needed for rerunning a failed build job # it's possible to rerun individual failed jobs when the build artifacts are available # if the artifacts have already been expired, the complete workflow can be rerun by closing and reopening the PR or by rebasing the PR diff --git a/.github/workflows/pulsar-ci.yaml b/.github/workflows/pulsar-ci.yaml index 721a1d2eafc72..f7dbc755264d6 100644 --- a/.github/workflows/pulsar-ci.yaml +++ b/.github/workflows/pulsar-ci.yaml @@ -36,7 +36,7 @@ concurrency: cancel-in-progress: true env: - MAVEN_OPTS: -Xss1500k -Xmx1024m -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 + MAVEN_OPTS: -Xss1500k -Xmx1024m -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.class=standard -Dmaven.wagon.http.retryHandler.count=3 -Dmaven.wagon.http.retryHandler.requestSentEnabled=true -Dmaven.wagon.http.serviceUnavailableRetryStrategy.class=standard -Dmaven.wagon.rto=60000 # defines the retention period for the intermediate build artifacts needed for rerunning a failed build job # it's possible to rerun individual failed jobs when the build artifacts are available # if the artifacts have already been expired, the complete workflow can be rerun by closing and reopening the PR or by rebasing the PR diff --git a/distribution/server/src/assemble/LICENSE.bin.txt b/distribution/server/src/assemble/LICENSE.bin.txt index 6b3455127b423..9205e072f7efe 100644 --- a/distribution/server/src/assemble/LICENSE.bin.txt +++ b/distribution/server/src/assemble/LICENSE.bin.txt @@ -343,35 +343,37 @@ The Apache Software License, Version 2.0 - org.apache.logging.log4j-log4j-slf4j-impl-2.18.0.jar - org.apache.logging.log4j-log4j-web-2.18.0.jar * Java Native Access JNA - - net.java.dev.jna-jna-5.12.1.jar - net.java.dev.jna-jna-jpms-5.12.1.jar - net.java.dev.jna-jna-platform-jpms-5.12.1.jar * BookKeeper - - org.apache.bookkeeper-bookkeeper-common-4.15.4.jar - - org.apache.bookkeeper-bookkeeper-common-allocator-4.15.4.jar - - org.apache.bookkeeper-bookkeeper-proto-4.15.4.jar - - org.apache.bookkeeper-bookkeeper-server-4.15.4.jar - - org.apache.bookkeeper-bookkeeper-tools-framework-4.15.4.jar - - org.apache.bookkeeper-circe-checksum-4.15.4.jar - - org.apache.bookkeeper-cpu-affinity-4.15.4.jar - - org.apache.bookkeeper-statelib-4.15.4.jar - - org.apache.bookkeeper-stream-storage-api-4.15.4.jar - - org.apache.bookkeeper-stream-storage-common-4.15.4.jar - - org.apache.bookkeeper-stream-storage-java-client-4.15.4.jar - - org.apache.bookkeeper-stream-storage-java-client-base-4.15.4.jar - - org.apache.bookkeeper-stream-storage-proto-4.15.4.jar - - org.apache.bookkeeper-stream-storage-server-4.15.4.jar - - org.apache.bookkeeper-stream-storage-service-api-4.15.4.jar - - org.apache.bookkeeper-stream-storage-service-impl-4.15.4.jar - - org.apache.bookkeeper.http-http-server-4.15.4.jar - - org.apache.bookkeeper.http-vertx-http-server-4.15.4.jar - - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.15.4.jar - - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.15.4.jar - - org.apache.distributedlog-distributedlog-common-4.15.4.jar - - org.apache.distributedlog-distributedlog-core-4.15.4-tests.jar - - org.apache.distributedlog-distributedlog-core-4.15.4.jar - - org.apache.distributedlog-distributedlog-protocol-4.15.4.jar - - org.apache.bookkeeper.stats-codahale-metrics-provider-4.15.4.jar + - org.apache.bookkeeper-bookkeeper-common-4.16.0.jar + - org.apache.bookkeeper-bookkeeper-common-allocator-4.16.0.jar + - org.apache.bookkeeper-bookkeeper-proto-4.16.0.jar + - org.apache.bookkeeper-bookkeeper-server-4.16.0.jar + - org.apache.bookkeeper-bookkeeper-tools-framework-4.16.0.jar + - org.apache.bookkeeper-circe-checksum-4.16.0.jar + - org.apache.bookkeeper-cpu-affinity-4.16.0.jar + - org.apache.bookkeeper-statelib-4.16.0.jar + - org.apache.bookkeeper-stream-storage-api-4.16.0.jar + - org.apache.bookkeeper-stream-storage-common-4.16.0.jar + - org.apache.bookkeeper-stream-storage-java-client-4.16.0.jar + - org.apache.bookkeeper-stream-storage-java-client-base-4.16.0.jar + - org.apache.bookkeeper-stream-storage-proto-4.16.0.jar + - org.apache.bookkeeper-stream-storage-server-4.16.0.jar + - org.apache.bookkeeper-stream-storage-service-api-4.16.0.jar + - org.apache.bookkeeper-stream-storage-service-impl-4.16.0.jar + - org.apache.bookkeeper.http-http-server-4.16.0.jar + - org.apache.bookkeeper.http-vertx-http-server-4.16.0.jar + - org.apache.bookkeeper.stats-bookkeeper-stats-api-4.16.0.jar + - org.apache.bookkeeper.stats-prometheus-metrics-provider-4.16.0.jar + - org.apache.distributedlog-distributedlog-common-4.16.0.jar + - org.apache.distributedlog-distributedlog-core-4.16.0-tests.jar + - org.apache.distributedlog-distributedlog-core-4.16.0.jar + - org.apache.distributedlog-distributedlog-protocol-4.16.0.jar + - org.apache.bookkeeper.stats-codahale-metrics-provider-4.16.0.jar + - org.apache.bookkeeper-bookkeeper-slogger-api-4.16.0.jar + - org.apache.bookkeeper-bookkeeper-slogger-slf4j-4.16.0.jar + - org.apache.bookkeeper-native-io-4.16.0.jar * Apache HTTP Client - org.apache.httpcomponents-httpclient-4.5.13.jar - org.apache.httpcomponents-httpcore-4.4.15.jar diff --git a/distribution/shell/src/assemble/LICENSE.bin.txt b/distribution/shell/src/assemble/LICENSE.bin.txt index 90896790b1fba..0ac08caa1c01d 100644 --- a/distribution/shell/src/assemble/LICENSE.bin.txt +++ b/distribution/shell/src/assemble/LICENSE.bin.txt @@ -390,9 +390,9 @@ The Apache Software License, Version 2.0 - log4j-web-2.18.0.jar * BookKeeper - - bookkeeper-common-allocator-4.15.4.jar - - cpu-affinity-4.15.4.jar - - circe-checksum-4.15.4.jar + - bookkeeper-common-allocator-4.16.0.jar + - cpu-affinity-4.16.0.jar + - circe-checksum-4.16.0.jar * AirCompressor - aircompressor-0.20.jar * AsyncHttpClient diff --git a/pom.xml b/pom.xml index 257018b0b0598..5367b211c6ef9 100644 --- a/pom.xml +++ b/pom.xml @@ -126,7 +126,7 @@ flexible messaging model and an intuitive client API. 1.21 - 4.15.4 + 4.16.0 3.8.1 1.5.0 1.10.0 @@ -149,7 +149,7 @@ flexible messaging model and an intuitive client API. 1.0.6 1.0.2.3 2.13.4.20221013 - 0.9.11 + 0.10.2 1.6.2 8.37 0.40.2 @@ -2455,5 +2455,13 @@ flexible messaging model and an intuitive client API. false + + bk-staging + bk-staging + https://repository.apache.org/content/repositories/orgapachebookkeeper-1082/ + + false + + diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java index 2f248a441cdee..3cc2da8db1e4d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java @@ -20,6 +20,7 @@ import com.google.common.annotations.Beta; import java.util.NavigableSet; +import java.util.concurrent.CompletableFuture; import org.apache.bookkeeper.mledger.impl.PositionImpl; /** @@ -81,8 +82,10 @@ public interface DelayedDeliveryTracker extends AutoCloseable { /** * Clear all delayed messages from the tracker. + * + * @return CompletableFuture */ - void clear(); + CompletableFuture clear(); /** * Close the subscription tracker and release all resources. 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 f55d5fd11694b..8de6ee58e2ce5 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 @@ -23,6 +23,7 @@ import java.time.Clock; import java.util.NavigableSet; import java.util.TreeSet; +import java.util.concurrent.CompletableFuture; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.impl.PositionImpl; @@ -147,8 +148,9 @@ public NavigableSet getScheduledMessages(int maxMessages) { } @Override - public void clear() { + public CompletableFuture clear() { this.priorityQueue.clear(); + return CompletableFuture.completedFuture(null); } @Override 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 index 7cfccff7ba328..5b7023be5034e 100644 --- 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 @@ -33,6 +33,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat; import org.apache.pulsar.common.util.Codec; +import org.apache.pulsar.common.util.FutureUtil; import org.roaringbitmap.RoaringBitmap; @Slf4j @@ -47,6 +48,9 @@ abstract class Bucket { protected final String dispatcherName; protected final ManagedCursor cursor; + + protected final FutureUtil.Sequencer sequencer; + protected final BucketSnapshotStorage bucketSnapshotStorage; long startLedgerId; @@ -67,9 +71,10 @@ abstract class Bucket { private volatile CompletableFuture snapshotCreateFuture; - Bucket(String dispatcherName, ManagedCursor cursor, + Bucket(String dispatcherName, ManagedCursor cursor, FutureUtil.Sequencer sequencer, BucketSnapshotStorage storage, long startLedgerId, long endLedgerId) { - this(dispatcherName, cursor, storage, startLedgerId, endLedgerId, new HashMap<>(), -1, -1, 0, 0, null, null); + this(dispatcherName, cursor, sequencer, storage, startLedgerId, endLedgerId, new HashMap<>(), -1, -1, 0, 0, + null, null); } boolean containsMessage(long ledgerId, long entryId) { @@ -154,12 +159,16 @@ CompletableFuture asyncSaveBucketSnapshot( private CompletableFuture putBucketKeyId(String bucketKey, Long bucketId) { Objects.requireNonNull(bucketId); - return executeWithRetry(() -> cursor.putCursorProperty(bucketKey, String.valueOf(bucketId)), - ManagedLedgerException.BadVersionException.class, MaxRetryTimes); + return sequencer.sequential(() -> { + return executeWithRetry(() -> cursor.putCursorProperty(bucketKey, String.valueOf(bucketId)), + ManagedLedgerException.BadVersionException.class, MaxRetryTimes); + }); } protected CompletableFuture removeBucketCursorProperty(String bucketKey) { - return executeWithRetry(() -> cursor.removeCursorProperty(bucketKey), - ManagedLedgerException.BadVersionException.class, MaxRetryTimes); + return sequencer.sequential(() -> { + return executeWithRetry(() -> cursor.removeCursorProperty(bucketKey), + 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 index 31fdaa6fb7667..39973928abf5d 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 @@ -31,6 +31,7 @@ import io.netty.util.Timeout; import io.netty.util.Timer; import java.time.Clock; +import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; import java.util.List; @@ -90,6 +91,8 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker private final Table snapshotSegmentLastIndexTable; + private static final Long INVALID_BUCKET_ID = -1L; + public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer, long tickTimeMillis, boolean isDelayedDeliveryDeliverAtTimeStrict, @@ -115,19 +118,23 @@ public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispat this.sharedBucketPriorityQueue = new TripleLongPriorityQueue(); this.immutableBuckets = TreeRangeMap.create(); this.snapshotSegmentLastIndexTable = HashBasedTable.create(); - this.lastMutableBucket = new MutableBucket(dispatcher.getName(), dispatcher.getCursor(), bucketSnapshotStorage); + this.lastMutableBucket = + new MutableBucket(dispatcher.getName(), dispatcher.getCursor(), FutureUtil.Sequencer.create(), + bucketSnapshotStorage); this.numberDelayedMessages = recoverBucketSnapshot(); } private synchronized long recoverBucketSnapshot() throws RuntimeException { ManagedCursor cursor = this.lastMutableBucket.getCursor(); + FutureUtil.Sequencer sequencer = this.lastMutableBucket.getSequencer(); Map, ImmutableBucket> toBeDeletedBucketMap = new HashMap<>(); cursor.getCursorProperties().keySet().forEach(key -> { if (key.startsWith(DELAYED_BUCKET_KEY_PREFIX)) { String[] keys = key.split(DELIMITER); checkArgument(keys.length == 3); ImmutableBucket immutableBucket = - new ImmutableBucket(dispatcher.getName(), cursor, this.lastMutableBucket.bucketSnapshotStorage, + new ImmutableBucket(dispatcher.getName(), cursor, sequencer, + this.lastMutableBucket.bucketSnapshotStorage, Long.parseLong(keys[1]), Long.parseLong(keys[2])); putAndCleanOverlapRange(Range.closed(immutableBucket.startLedgerId, immutableBucket.endLedgerId), immutableBucket, toBeDeletedBucketMap); @@ -246,12 +253,12 @@ private void afterCreateImmutableBucket(Pair immu immutableBucket); immutableBucket.getSnapshotCreateFuture().ifPresent(createFuture -> { - CompletableFuture future = createFuture.whenComplete((__, ex) -> { + CompletableFuture future = createFuture.handle((bucketId, ex) -> { if (ex == null) { immutableBucket.setSnapshotSegments(null); log.info("[{}] Creat bucket snapshot finish, bucketKey: {}", dispatcher.getName(), immutableBucket.bucketKey()); - return; + return bucketId; } //TODO Record create snapshot failed @@ -272,11 +279,12 @@ private void afterCreateImmutableBucket(Pair immu }); immutableBucket.setCurrentSegmentEntryId(immutableBucket.lastSegmentEntryId); - immutableBuckets.remove( + immutableBuckets.asMapOfRanges().remove( Range.closed(immutableBucket.startLedgerId, immutableBucket.endLedgerId)); snapshotSegmentLastIndexTable.remove(lastDelayedIndex.getLedgerId(), lastDelayedIndex.getTimestamp()); } + return INVALID_BUCKET_ID; }); immutableBucket.setSnapshotCreateFuture(future); }); @@ -308,12 +316,7 @@ public synchronized boolean addMessage(long ledgerId, long entryId, long deliver lastMutableBucket.resetLastMutableBucketRange(); if (immutableBuckets.asMapOfRanges().size() > maxNumBuckets) { - try { - asyncMergeBucketSnapshot().get(2 * AsyncOperationTimeoutSeconds * MaxRetryTimes, TimeUnit.SECONDS); - } catch (Exception e) { - // Ignore exception to merge bucket on the next schedule. - log.error("[{}] An exception occurs when merge bucket snapshot.", dispatcher.getName(), e); - } + asyncMergeBucketSnapshot(); } } @@ -341,18 +344,26 @@ public synchronized boolean addMessage(long ledgerId, long entryId, long deliver private synchronized CompletableFuture asyncMergeBucketSnapshot() { List values = immutableBuckets.asMapOfRanges().values().stream().toList(); long minNumberMessages = Long.MAX_VALUE; + long minScheduleTimestamp = Long.MAX_VALUE; int minIndex = -1; for (int i = 0; i + 1 < values.size(); i++) { ImmutableBucket bucketL = values.get(i); ImmutableBucket bucketR = values.get(i + 1); - long numberMessages = bucketL.numberBucketDelayedMessages + bucketR.numberBucketDelayedMessages; - if (numberMessages < minNumberMessages) { - minNumberMessages = (int) numberMessages; - if (bucketL.lastSegmentEntryId > bucketL.getCurrentSegmentEntryId() - && bucketR.lastSegmentEntryId > bucketR.getCurrentSegmentEntryId() - && bucketL.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE).isDone() - && bucketR.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE).isDone()) { - minIndex = i; + // We should skip the bucket which last segment already been load to memory, avoid record replicated index. + if (bucketL.lastSegmentEntryId > bucketL.getCurrentSegmentEntryId() + && bucketR.lastSegmentEntryId > bucketR.getCurrentSegmentEntryId() + // Skip the bucket that is merging + && !bucketL.merging && !bucketR.merging){ + long scheduleTimestamp = + Math.min(bucketL.firstScheduleTimestamps.get(bucketL.currentSegmentEntryId + 1), + bucketR.firstScheduleTimestamps.get(bucketR.currentSegmentEntryId + 1)); + long numberMessages = bucketL.numberBucketDelayedMessages + bucketR.numberBucketDelayedMessages; + if (scheduleTimestamp <= minScheduleTimestamp) { + minScheduleTimestamp = scheduleTimestamp; + if (numberMessages < minNumberMessages) { + minNumberMessages = numberMessages; + minIndex = i; + } } } } @@ -369,7 +380,14 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot() { log.info("[{}] Merging bucket snapshot, bucketAKey: {}, bucketBKey: {}", dispatcher.getName(), immutableBucketA.bucketKey(), immutableBucketB.bucketKey()); } + + immutableBucketA.merging = true; + immutableBucketB.merging = true; return asyncMergeBucketSnapshot(immutableBucketA, immutableBucketB).whenComplete((__, ex) -> { + synchronized (this) { + immutableBucketA.merging = false; + immutableBucketB.merging = false; + } if (ex != null) { log.error("[{}] Failed to merge bucket snapshot, bucketAKey: {}, bucketBKey: {}", dispatcher.getName(), immutableBucketA.bucketKey(), immutableBucketB.bucketKey(), ex); @@ -382,46 +400,59 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot() { private synchronized CompletableFuture asyncMergeBucketSnapshot(ImmutableBucket bucketA, ImmutableBucket bucketB) { - CompletableFuture> futureA = - bucketA.getRemainSnapshotSegment(); - CompletableFuture> futureB = - bucketB.getRemainSnapshotSegment(); - return futureA.thenCombine(futureB, CombinedSegmentDelayedIndexQueue::wrap) - .thenAccept(combinedDelayedIndexQueue -> { - Pair immutableBucketDelayedIndexPair = - lastMutableBucket.createImmutableBucketAndAsyncPersistent( - timeStepPerBucketSnapshotSegmentInMillis, maxIndexesPerBucketSnapshotSegment, - sharedBucketPriorityQueue, combinedDelayedIndexQueue, bucketA.startLedgerId, - bucketB.endLedgerId); - - // Merge bit map to new bucket - Map delayedIndexBitMapA = bucketA.getDelayedIndexBitMap(); - Map delayedIndexBitMapB = bucketB.getDelayedIndexBitMap(); - Map delayedIndexBitMap = new HashMap<>(delayedIndexBitMapA); - delayedIndexBitMapB.forEach((ledgerId, bitMapB) -> { - delayedIndexBitMap.compute(ledgerId, (k, bitMapA) -> { - if (bitMapA == null) { - return bitMapB; - } - - bitMapA.or(bitMapB); - return bitMapA; - }); - }); - immutableBucketDelayedIndexPair.getLeft().setDelayedIndexBitMap(delayedIndexBitMap); + CompletableFuture createAFuture = bucketA.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE); + CompletableFuture createBFuture = bucketB.getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE); - afterCreateImmutableBucket(immutableBucketDelayedIndexPair); + return CompletableFuture.allOf(createAFuture, createBFuture).thenCompose(bucketId -> { + if (INVALID_BUCKET_ID.equals(createAFuture.join()) || INVALID_BUCKET_ID.equals(createBFuture.join())) { + return FutureUtil.failedFuture(new RuntimeException("Can't merge buckets due to bucket create failed")); + } - immutableBucketDelayedIndexPair.getLeft().getSnapshotCreateFuture() - .orElse(NULL_LONG_PROMISE).thenCompose(___ -> { - CompletableFuture removeAFuture = bucketA.asyncDeleteBucketSnapshot(); - CompletableFuture removeBFuture = bucketB.asyncDeleteBucketSnapshot(); - return CompletableFuture.allOf(removeAFuture, removeBFuture); + CompletableFuture> futureA = + bucketA.getRemainSnapshotSegment(); + CompletableFuture> futureB = + bucketB.getRemainSnapshotSegment(); + return futureA.thenCombine(futureB, CombinedSegmentDelayedIndexQueue::wrap) + .thenAccept(combinedDelayedIndexQueue -> { + synchronized (BucketDelayedDeliveryTracker.this) { + Pair immutableBucketDelayedIndexPair = + lastMutableBucket.createImmutableBucketAndAsyncPersistent( + timeStepPerBucketSnapshotSegmentInMillis, + maxIndexesPerBucketSnapshotSegment, + sharedBucketPriorityQueue, combinedDelayedIndexQueue, bucketA.startLedgerId, + bucketB.endLedgerId); + + // Merge bit map to new bucket + Map delayedIndexBitMapA = bucketA.getDelayedIndexBitMap(); + Map delayedIndexBitMapB = bucketB.getDelayedIndexBitMap(); + Map delayedIndexBitMap = new HashMap<>(delayedIndexBitMapA); + delayedIndexBitMapB.forEach((ledgerId, bitMapB) -> { + delayedIndexBitMap.compute(ledgerId, (k, bitMapA) -> { + if (bitMapA == null) { + return bitMapB; + } + + bitMapA.or(bitMapB); + return bitMapA; + }); + }); + immutableBucketDelayedIndexPair.getLeft().setDelayedIndexBitMap(delayedIndexBitMap); + + afterCreateImmutableBucket(immutableBucketDelayedIndexPair); + + immutableBucketDelayedIndexPair.getLeft().getSnapshotCreateFuture() + .orElse(NULL_LONG_PROMISE).thenCompose(___ -> { + CompletableFuture removeAFuture = bucketA.asyncDeleteBucketSnapshot(); + CompletableFuture removeBFuture = bucketB.asyncDeleteBucketSnapshot(); + return CompletableFuture.allOf(removeAFuture, removeBFuture); + }); + + Map, ImmutableBucket> immutableBucketMap = immutableBuckets.asMapOfRanges(); + immutableBucketMap.remove(Range.closed(bucketA.startLedgerId, bucketA.endLedgerId)); + immutableBucketMap.remove(Range.closed(bucketB.startLedgerId, bucketB.endLedgerId)); + } }); - - immutableBuckets.remove(Range.closed(bucketA.startLedgerId, bucketA.endLedgerId)); - immutableBuckets.remove(Range.closed(bucketB.startLedgerId, bucketB.endLedgerId)); - }); + }); } @Override @@ -477,6 +508,12 @@ public synchronized NavigableSet getScheduledMessages(int maxMessa ImmutableBucket bucket = snapshotSegmentLastIndexTable.get(ledgerId, entryId); if (bucket != null && immutableBuckets.asMapOfRanges().containsValue(bucket)) { + if (bucket.merging) { + log.info("[{}] Skip load to wait for bucket snapshot merge finish, bucketKey:{}", + dispatcher.getName(), bucket.bucketKey()); + break; + } + final int preSegmentEntryId = bucket.currentSegmentEntryId; if (log.isDebugEnabled()) { log.debug("[{}] Loading next bucket snapshot segment, bucketKey: {}, nextSegmentEntryId: {}", @@ -494,14 +531,15 @@ public synchronized NavigableSet getScheduledMessages(int maxMessa } if (bucket.currentSegmentEntryId == bucket.lastSegmentEntryId) { - immutableBuckets.remove(Range.closed(bucket.startLedgerId, bucket.endLedgerId)); + immutableBuckets.asMapOfRanges().remove(Range.closed(bucket.startLedgerId, bucket.endLedgerId)); bucket.asyncDeleteBucketSnapshot(); continue; } bucket.asyncLoadNextBucketSnapshotEntry().thenAccept(indexList -> { if (CollectionUtils.isEmpty(indexList)) { - immutableBuckets.remove(Range.closed(bucket.startLedgerId, bucket.endLedgerId)); + immutableBuckets.asMapOfRanges() + .remove(Range.closed(bucket.startLedgerId, bucket.endLedgerId)); bucket.asyncDeleteBucketSnapshot(); return; } @@ -518,14 +556,13 @@ public synchronized NavigableSet getScheduledMessages(int maxMessa // Back bucket state bucket.setCurrentSegmentEntryId(preSegmentEntryId); - log.error("[{}] Failed to load bucket snapshot segment, bucketKey: {}", - dispatcher.getName(), bucket.bucketKey(), ex); + log.error("[{}] Failed to load bucket snapshot segment, bucketKey: {}, segmentEntryId: {}", + dispatcher.getName(), bucket.bucketKey(), preSegmentEntryId + 1, ex); } else { log.info("[{}] Load next bucket snapshot segment finish, bucketKey: {}, segmentEntryId: {}", - dispatcher.getName(), bucket.bucketKey(), bucket.currentSegmentEntryId); + dispatcher.getName(), bucket.bucketKey(), preSegmentEntryId + 1); } - }).get(AsyncOperationTimeoutSeconds * MaxRetryTimes, TimeUnit.SECONDS); - snapshotSegmentLastIndexTable.remove(ledgerId, entryId); + }).get(AsyncOperationTimeoutSeconds * (MaxRetryTimes + 1), TimeUnit.SECONDS); } catch (Exception e) { // Ignore exception to reload this segment on the next schedule. log.error("[{}] An exception occurs when load next bucket snapshot, bucketKey:{}", @@ -533,6 +570,7 @@ public synchronized NavigableSet getScheduledMessages(int maxMessa break; } } + snapshotSegmentLastIndexTable.remove(ledgerId, entryId); positions.add(new PositionImpl(ledgerId, entryId)); @@ -554,30 +592,46 @@ public boolean shouldPauseAllDeliveries() { } @Override - public synchronized void clear() { - cleanImmutableBuckets(true); - sharedBucketPriorityQueue.clear(); - lastMutableBucket.clear(); - snapshotSegmentLastIndexTable.clear(); - numberDelayedMessages = 0; + public synchronized CompletableFuture clear() { + return cleanImmutableBuckets(true).thenRun(() -> { + synchronized (this) { + sharedBucketPriorityQueue.clear(); + lastMutableBucket.clear(); + snapshotSegmentLastIndexTable.clear(); + numberDelayedMessages = 0; + } + }); } @Override public synchronized void close() { super.close(); lastMutableBucket.close(); - cleanImmutableBuckets(false); sharedBucketPriorityQueue.close(); + try { + cleanImmutableBuckets(false).get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS); + } catch (Exception e) { + log.warn("[{}] Failed wait to snapshot generate", dispatcher.getName(), e); + } } - private void cleanImmutableBuckets(boolean delete) { + private CompletableFuture cleanImmutableBuckets(boolean delete) { if (immutableBuckets != null) { + List> futures = new ArrayList<>(); Iterator iterator = immutableBuckets.asMapOfRanges().values().iterator(); while (iterator.hasNext()) { ImmutableBucket bucket = iterator.next(); - bucket.clear(delete); + if (delete) { + futures.add(bucket.clear()); + } else { + bucket.getSnapshotCreateFuture().ifPresent(future -> futures.add(future.thenApply(x -> null))); + } + numberDelayedMessages -= bucket.getNumberBucketDelayedMessages(); iterator.remove(); } + return FutureUtil.waitForAll(futures); + } else { + return CompletableFuture.completedFuture(null); } } 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 c9223efa09243..969d326e28187 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 @@ -19,15 +19,14 @@ package org.apache.pulsar.broker.delayed.bucket; import static org.apache.bookkeeper.mledger.util.Futures.executeWithRetry; -import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.AsyncOperationTimeoutSeconds; import static org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker.NULL_LONG_PROMISE; import com.google.protobuf.ByteString; +import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; import java.util.function.Supplier; import lombok.Setter; import lombok.extern.slf4j.Slf4j; @@ -37,6 +36,7 @@ import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat; import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.DelayedIndex; import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata; +import org.apache.pulsar.common.util.FutureUtil; import org.roaringbitmap.RoaringBitmap; import org.roaringbitmap.buffer.ImmutableRoaringBitmap; @@ -44,11 +44,16 @@ class ImmutableBucket extends Bucket { @Setter - private volatile List snapshotSegments; + private List snapshotSegments; - ImmutableBucket(String dispatcherName, ManagedCursor cursor, + boolean merging = false; + + @Setter + List firstScheduleTimestamps = new ArrayList<>(); + + ImmutableBucket(String dispatcherName, ManagedCursor cursor, FutureUtil.Sequencer sequencer, BucketSnapshotStorage storage, long startLedgerId, long endLedgerId) { - super(dispatcherName, cursor, storage, startLedgerId, endLedgerId); + super(dispatcherName, cursor, sequencer, storage, startLedgerId, endLedgerId); } public Optional> getSnapshotSegments() { @@ -92,6 +97,9 @@ private CompletableFuture> asyncLoadNextBucketSnapshotEntry(b this.setLastSegmentEntryId(metadataList.size()); this.recoverDelayedIndexBitMapAndNumber(nextSnapshotEntryIndex, metadataList); + List firstScheduleTimestamps = metadataList.stream().map( + SnapshotSegmentMetadata::getMinScheduleTimestamp).toList(); + this.setFirstScheduleTimestamps(firstScheduleTimestamps); return nextSnapshotEntryIndex + 1; }); @@ -108,8 +116,9 @@ private CompletableFuture> asyncLoadNextBucketSnapshotEntry(b () -> bucketSnapshotStorage.getBucketSnapshotSegment(bucketId, nextSegmentEntryId, nextSegmentEntryId).whenComplete((___, ex) -> { if (ex != null) { - log.warn("[{}] Failed to get bucket snapshot segment. bucketKey: {}, bucketId: {}", - dispatcherName, bucketKey(), bucketId, ex); + log.warn("[{}] Failed to get bucket snapshot segment. bucketKey: {}," + + " bucketId: {}, segmentEntryId: {}", dispatcherName, bucketKey(), + bucketId, nextSegmentEntryId, ex); } }), BucketSnapshotPersistenceException.class, MaxRetryTimes) .thenApply(bucketSnapshotSegments -> { @@ -157,8 +166,10 @@ CompletableFuture> return bucketSnapshotStorage.getBucketSnapshotSegment(getAndUpdateBucketId(), nextSegmentEntryId, lastSegmentEntryId).whenComplete((__, ex) -> { if (ex != null) { - log.warn("[{}] Failed to get remain bucket snapshot segment, bucketKey: {}.", - dispatcherName, bucketKey(), ex); + log.warn( + "[{}] Failed to get remain bucket snapshot segment, bucketKey: {}," + + " nextSegmentEntryId: {}, lastSegmentEntryId: {}", + dispatcherName, bucketKey(), nextSegmentEntryId, lastSegmentEntryId, ex); } }); }, BucketSnapshotPersistenceException.class, MaxRetryTimes); @@ -180,48 +191,9 @@ CompletableFuture asyncDeleteBucketSnapshot() { }); } - void clear(boolean delete) { + CompletableFuture clear() { delayedIndexBitMap.clear(); - if (delete) { - final String bucketKey = bucketKey(); - try { - getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE).exceptionally(e -> null).thenCompose(__ -> { - if (getSnapshotCreateFuture().isPresent() && getBucketId().isEmpty()) { - log.error("[{}] Can't found bucketId, don't execute delete operate, bucketKey: {}", - dispatcherName, bucketKey); - return CompletableFuture.completedFuture(null); - } - long bucketId = getAndUpdateBucketId(); - return removeBucketCursorProperty(bucketKey()).thenAccept(___ -> { - executeWithRetry(() -> bucketSnapshotStorage.deleteBucketSnapshot(bucketId), - BucketSnapshotPersistenceException.class, MaxRetryTimes) - .whenComplete((____, ex) -> { - if (ex != null) { - log.error("[{}] Failed to delete bucket snapshot, bucketId: {}, bucketKey: {}", - dispatcherName, bucketId, bucketKey, ex); - } else { - log.info("[{}] Delete bucket snapshot finish, bucketId: {}, bucketKey: {}", - dispatcherName, bucketId, bucketKey); - } - }); - }); - }).get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS); - } catch (Exception e) { - log.error("Failed to clear bucket snapshot, bucketKey: {}", bucketKey, e); - if (e instanceof InterruptedException) { - Thread.currentThread().interrupt(); - } - throw new RuntimeException(e); - } - } else { - getSnapshotCreateFuture().ifPresent(snapshotGenerateFuture -> { - try { - snapshotGenerateFuture.get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS); - } catch (Exception e) { - log.warn("[{}] Failed wait to snapshot generate, bucketId: {}, bucketKey: {}", dispatcherName, - getBucketId(), bucketKey()); - } - }); - } + return getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE).exceptionally(e -> null) + .thenCompose(__ -> asyncDeleteBucketSnapshot()); } } 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 e743f39e6920d..f8a4ecc7a4ddb 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 @@ -34,6 +34,7 @@ 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.FutureUtil; import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue; import org.roaringbitmap.RoaringBitmap; @@ -42,9 +43,9 @@ class MutableBucket extends Bucket implements AutoCloseable { private final TripleLongPriorityQueue priorityQueue; - MutableBucket(String dispatcherName, ManagedCursor cursor, + MutableBucket(String dispatcherName, ManagedCursor cursor, FutureUtil.Sequencer sequencer, BucketSnapshotStorage bucketSnapshotStorage) { - super(dispatcherName, cursor, bucketSnapshotStorage, -1L, -1L); + super(dispatcherName, cursor, sequencer, bucketSnapshotStorage, -1L, -1L); this.priorityQueue = new TripleLongPriorityQueue(); } @@ -75,11 +76,15 @@ Pair createImmutableBucketAndAsyncPersistent( SnapshotSegment.Builder snapshotSegmentBuilder = SnapshotSegment.newBuilder(); SnapshotSegmentMetadata.Builder segmentMetadataBuilder = SnapshotSegmentMetadata.newBuilder(); + List firstScheduleTimestamps = new ArrayList<>(); long currentTimestampUpperLimit = 0; + long currentFirstTimestamp = 0L; while (!delayedIndexQueue.isEmpty()) { DelayedIndex delayedIndex = delayedIndexQueue.peek(); long timestamp = delayedIndex.getTimestamp(); if (currentTimestampUpperLimit == 0) { + currentFirstTimestamp = timestamp; + firstScheduleTimestamps.add(currentFirstTimestamp); currentTimestampUpperLimit = timestamp + timeStepPerBucketSnapshotSegment - 1; } @@ -104,6 +109,7 @@ Pair createImmutableBucketAndAsyncPersistent( || (maxIndexesPerBucketSnapshotSegment != -1 && snapshotSegmentBuilder.getIndexesCount() >= maxIndexesPerBucketSnapshotSegment)) { segmentMetadataBuilder.setMaxScheduleTimestamp(timestamp); + segmentMetadataBuilder.setMinScheduleTimestamp(currentFirstTimestamp); currentTimestampUpperLimit = 0; Iterator> iterator = bitMap.entrySet().iterator(); @@ -129,11 +135,12 @@ Pair createImmutableBucketAndAsyncPersistent( final int lastSegmentEntryId = segmentMetadataList.size(); - ImmutableBucket bucket = new ImmutableBucket(dispatcherName, cursor, bucketSnapshotStorage, + ImmutableBucket bucket = new ImmutableBucket(dispatcherName, cursor, sequencer, bucketSnapshotStorage, startLedgerId, endLedgerId); bucket.setCurrentSegmentEntryId(1); bucket.setNumberBucketDelayedMessages(numMessages); bucket.setLastSegmentEntryId(lastSegmentEntryId); + bucket.setFirstScheduleTimestamps(firstScheduleTimestamps); // Skip first segment, because it has already been loaded List snapshotSegments = bucketSnapshotSegments.subList(1, bucketSnapshotSegments.size()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java index 48877ce53f801..9b0c4e885e64d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java @@ -110,8 +110,8 @@ default long getNumberOfDelayedMessages() { return 0; } - default void clearDelayedMessages() { - //No-op + default CompletableFuture clearDelayedMessages() { + return CompletableFuture.completedFuture(null); } default void cursorIsReset() { 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 059820b1b66c3..e5c9e85bac3f2 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 @@ -48,6 +48,7 @@ import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.delayed.AbstractDelayedDeliveryTracker; +import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory; import org.apache.pulsar.broker.delayed.DelayedDeliveryTracker; import org.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTracker; import org.apache.pulsar.broker.delayed.bucket.BucketDelayedDeliveryTracker; @@ -1089,8 +1090,26 @@ public synchronized long getNumberOfDelayedMessages() { } @Override - public void clearDelayedMessages() { - this.delayedDeliveryTracker.ifPresent(DelayedDeliveryTracker::clear); + public CompletableFuture clearDelayedMessages() { + if (!topic.isDelayedDeliveryEnabled()) { + return CompletableFuture.completedFuture(null); + } + + if (delayedDeliveryTracker.isEmpty() && topic.getBrokerService() + .getDelayedDeliveryTrackerFactory() instanceof BucketDelayedDeliveryTrackerFactory) { + synchronized (this) { + if (delayedDeliveryTracker.isEmpty()) { + delayedDeliveryTracker = Optional + .of(topic.getBrokerService().getDelayedDeliveryTrackerFactory().newTracker(this)); + } + } + } + + if (delayedDeliveryTracker.isPresent()) { + return this.delayedDeliveryTracker.get().clear(); + } else { + return CompletableFuture.completedFuture(null); + } } @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index e07d7bee500a7..bdb3c9fc391ce 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -649,9 +649,16 @@ public void clearBacklogComplete(Object ctx) { cursor.getNumberOfEntriesInBacklog(false)); } if (dispatcher != null) { - dispatcher.clearDelayedMessages(); + dispatcher.clearDelayedMessages().whenComplete((__, ex) -> { + if (ex != null) { + future.completeExceptionally(ex); + } else { + future.complete(null); + } + }); + } else { + future.complete(null); } - future.complete(null); } @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index b3b6526eea54d..82a4f5312357a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -79,6 +79,7 @@ import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.PulsarServerException; +import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory; import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl; import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateCompactionStrategy; import org.apache.pulsar.broker.namespace.NamespaceService; @@ -1078,13 +1079,13 @@ public CompletableFuture unsubscribe(String subscriptionName) { new AsyncCallbacks.DeleteLedgerCallback() { @Override public void deleteLedgerComplete(Object ctx) { - asyncDeleteCursor(subscriptionName, unsubscribeFuture); + asyncDeleteCursorWithClearDelayedMessage(subscriptionName, unsubscribeFuture); } @Override public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) { if (exception instanceof MetadataNotFoundException) { - asyncDeleteCursor(subscriptionName, unsubscribeFuture); + asyncDeleteCursorWithClearDelayedMessage(subscriptionName, unsubscribeFuture); return; } @@ -1094,12 +1095,51 @@ public void deleteLedgerFailed(ManagedLedgerException exception, Object ctx) { } }, null); } else { - asyncDeleteCursor(subscriptionName, unsubscribeFuture); + asyncDeleteCursorWithClearDelayedMessage(subscriptionName, unsubscribeFuture); } return unsubscribeFuture; } + private void asyncDeleteCursorWithClearDelayedMessage(String subscriptionName, + CompletableFuture unsubscribeFuture) { + if (!isDelayedDeliveryEnabled() + || !(brokerService.getDelayedDeliveryTrackerFactory() instanceof BucketDelayedDeliveryTrackerFactory)) { + asyncDeleteCursor(subscriptionName, unsubscribeFuture); + return; + } + + PersistentSubscription persistentSubscription = subscriptions.get(subscriptionName); + if (persistentSubscription == null) { + log.warn("[{}][{}] Can't find subscription, skip clear delayed message", topic, subscriptionName); + asyncDeleteCursor(subscriptionName, unsubscribeFuture); + return; + } + + Dispatcher dispatcher = persistentSubscription.getDispatcher(); + final Dispatcher temporaryDispatcher; + if (dispatcher == null) { + log.info("[{}][{}] Dispatcher is null, try to create temporary dispatcher to clear delayed message", topic, + subscriptionName); + dispatcher = temporaryDispatcher = + new PersistentDispatcherMultipleConsumers(this, persistentSubscription.cursor, + persistentSubscription); + } else { + temporaryDispatcher = null; + } + + dispatcher.clearDelayedMessages().whenComplete((__, ex) -> { + if (ex != null) { + unsubscribeFuture.completeExceptionally(ex); + } else { + asyncDeleteCursor(subscriptionName, unsubscribeFuture); + } + if (temporaryDispatcher != null) { + temporaryDispatcher.close(); + } + }); + } + private void asyncDeleteCursor(String subscriptionName, CompletableFuture unsubscribeFuture) { ledger.asyncDeleteCursor(Codec.encode(subscriptionName), new DeleteCursorCallback() { @Override diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/LongAdderCounter.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/LongAdderCounter.java index ff9d9302456d1..8ade2bc883f9a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/LongAdderCounter.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/LongAdderCounter.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.stats.prometheus.metrics; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.LongAdder; import org.apache.bookkeeper.stats.Counter; @@ -50,10 +51,16 @@ public void dec() { } @Override - public void add(long delta) { + public void addCount(long delta) { counter.add(delta); } + @Override + public void addLatency(long eventLatency, TimeUnit unit) { + long valueMillis = unit.toMillis(eventLatency); + counter.add(valueMillis); + } + @Override public Long get() { return counter.sum(); diff --git a/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto b/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto index 8414a583fe5b0..6996b860c5249 100644 --- a/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto +++ b/pulsar-broker/src/main/proto/DelayedMessageIndexBucketSnapshotFormat.proto @@ -31,6 +31,7 @@ message DelayedIndex { message SnapshotSegmentMetadata { map delayed_index_bit_map = 1; required uint64 max_schedule_timestamp = 2; + required uint64 min_schedule_timestamp = 3; } message SnapshotSegment { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/SameThreadOrderedSafeExecutor.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/SameThreadOrderedSafeExecutor.java index 5f6a32a61bd25..258188a31f5d0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/SameThreadOrderedSafeExecutor.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/SameThreadOrderedSafeExecutor.java @@ -21,7 +21,6 @@ import io.netty.util.concurrent.DefaultThreadFactory; import org.apache.bookkeeper.common.util.OrderedExecutor; -import org.apache.bookkeeper.common.util.SafeRunnable; import org.apache.bookkeeper.stats.NullStatsLogger; public class SameThreadOrderedSafeExecutor extends OrderedExecutor { @@ -46,12 +45,12 @@ public void execute(Runnable r) { } @Override - public void executeOrdered(int orderingKey, SafeRunnable r) { + public void executeOrdered(int orderingKey, Runnable r) { r.run(); } @Override - public void executeOrdered(long orderingKey, SafeRunnable r) { + public void executeOrdered(long orderingKey, Runnable r) { r.run(); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java index 72052d22b859d..a628b58e10d32 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/BookkeeperBucketSnapshotStorageTest.java @@ -73,6 +73,7 @@ public void testCreateSnapshot() throws ExecutionException, InterruptedException public void testGetSnapshot() throws ExecutionException, InterruptedException { DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata segmentMetadata = DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata.newBuilder() + .setMinScheduleTimestamp(System.currentTimeMillis()) .setMaxScheduleTimestamp(System.currentTimeMillis()) .putDelayedIndexBitMap(100L, ByteString.copyFrom(new byte[1])).build(); @@ -122,6 +123,7 @@ public void testGetSnapshotMetadata() throws ExecutionException, InterruptedExce DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata segmentMetadata = DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata.newBuilder() .setMaxScheduleTimestamp(timeMillis) + .setMinScheduleTimestamp(timeMillis) .putAllDelayedIndexBitMap(map).build(); DelayedMessageIndexBucketSnapshotFormat.SnapshotMetadata snapshotMetadata = @@ -172,6 +174,7 @@ public void testDeleteSnapshot() throws ExecutionException, InterruptedException public void testGetBucketSnapshotLength() throws ExecutionException, InterruptedException { DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata segmentMetadata = DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata.newBuilder() + .setMinScheduleTimestamp(System.currentTimeMillis()) .setMaxScheduleTimestamp(System.currentTimeMillis()) .putDelayedIndexBitMap(100L, ByteString.copyFrom(new byte[1])).build(); 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 74101f00b960c..fbb866d48ed69 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 @@ -39,7 +39,9 @@ import java.util.NavigableSet; import java.util.Set; import java.util.TreeMap; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.bookkeeper.mledger.impl.PositionImpl; @@ -52,6 +54,7 @@ import org.roaringbitmap.buffer.ImmutableRoaringBitmap; import org.testcontainers.shaded.org.apache.commons.lang3.mutable.MutableLong; import org.testcontainers.shaded.org.awaitility.Awaitility; +import org.testng.Assert; import org.testng.annotations.AfterMethod; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @@ -251,18 +254,28 @@ public void testRoaringBitmapSerialize() { } @Test(dataProvider = "delayedTracker") - public void testMergeSnapshot(BucketDelayedDeliveryTracker tracker) { + public void testMergeSnapshot(final BucketDelayedDeliveryTracker tracker) { for (int i = 1; i <= 110; i++) { tracker.addMessage(i, i, i * 10); + Awaitility.await().untilAsserted(() -> { + Assert.assertTrue( + tracker.getImmutableBuckets().asMapOfRanges().values().stream().noneMatch(x -> x.merging)); + }); } assertEquals(110, tracker.getNumberOfDelayedMessages()); int size = tracker.getImmutableBuckets().asMapOfRanges().size(); - assertEquals(10, size); + Awaitility.await().untilAsserted(() -> { + assertEquals(10, size); + }); tracker.addMessage(111, 1011, 111 * 10); + Awaitility.await().untilAsserted(() -> { + Assert.assertTrue( + tracker.getImmutableBuckets().asMapOfRanges().values().stream().noneMatch(x -> x.merging)); + }); MutableLong delayedMessagesInSnapshot = new MutableLong(); tracker.getImmutableBuckets().asMapOfRanges().forEach((k, v) -> { @@ -271,26 +284,28 @@ public void testMergeSnapshot(BucketDelayedDeliveryTracker tracker) { tracker.close(); - tracker = new BucketDelayedDeliveryTracker(dispatcher, timer, 1000, clock, - true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1,10); + BucketDelayedDeliveryTracker tracker2 = new BucketDelayedDeliveryTracker(dispatcher, timer, 1000, clock, + true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 10); - assertEquals(tracker.getNumberOfDelayedMessages(), delayedMessagesInSnapshot.getValue()); + assertEquals(tracker2.getNumberOfDelayedMessages(), delayedMessagesInSnapshot.getValue()); for (int i = 1; i <= 110; i++) { - tracker.addMessage(i, i, i * 10); + tracker2.addMessage(i, i, i * 10); } clockTime.set(110 * 10); - NavigableSet scheduledMessages = tracker.getScheduledMessages(110); + NavigableSet scheduledMessages = tracker2.getScheduledMessages(110); for (int i = 1; i <= 110; i++) { PositionImpl position = scheduledMessages.pollFirst(); assertEquals(position, PositionImpl.get(i, i)); } + + tracker2.close(); } @Test(dataProvider = "delayedTracker") - public void testWithBkException(BucketDelayedDeliveryTracker tracker) { + public void testWithBkException(final BucketDelayedDeliveryTracker tracker) { MockBucketSnapshotStorage mockBucketSnapshotStorage = (MockBucketSnapshotStorage) bucketSnapshotStorage; mockBucketSnapshotStorage.injectCreateException( new BucketSnapshotPersistenceException("Bookie operation timeout, op: Create entry")); @@ -308,11 +323,25 @@ public void testWithBkException(BucketDelayedDeliveryTracker tracker) { for (int i = 1; i <= 110; i++) { tracker.addMessage(i, i, i * 10); + Awaitility.await().untilAsserted(() -> { + Assert.assertTrue( + tracker.getImmutableBuckets().asMapOfRanges().values().stream().noneMatch(x -> x.merging)); + }); } assertEquals(110, tracker.getNumberOfDelayedMessages()); + int size = tracker.getImmutableBuckets().asMapOfRanges().size(); + + Awaitility.await().untilAsserted(() -> { + assertEquals(10, size); + }); + tracker.addMessage(111, 1011, 111 * 10); + Awaitility.await().untilAsserted(() -> { + Assert.assertTrue( + tracker.getImmutableBuckets().asMapOfRanges().values().stream().noneMatch(x -> x.merging)); + }); MutableLong delayedMessagesInSnapshot = new MutableLong(); tracker.getImmutableBuckets().asMapOfRanges().forEach((k, v) -> { @@ -321,11 +350,11 @@ public void testWithBkException(BucketDelayedDeliveryTracker tracker) { tracker.close(); - tracker = new BucketDelayedDeliveryTracker(dispatcher, timer, 1000, clock, + BucketDelayedDeliveryTracker tracker2 = new BucketDelayedDeliveryTracker(dispatcher, timer, 1000, clock, true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1,10); Long delayedMessagesInSnapshotValue = delayedMessagesInSnapshot.getValue(); - assertEquals(tracker.getNumberOfDelayedMessages(), delayedMessagesInSnapshotValue); + assertEquals(tracker2.getNumberOfDelayedMessages(), delayedMessagesInSnapshotValue); clockTime.set(110 * 10); @@ -338,14 +367,16 @@ public void testWithBkException(BucketDelayedDeliveryTracker tracker) { mockBucketSnapshotStorage.injectGetSegmentException( new BucketSnapshotPersistenceException("Bookie operation timeout4, op: Get entry")); - assertEquals(tracker.getScheduledMessages(100).size(), 0); + assertEquals(tracker2.getScheduledMessages(100).size(), 0); - assertEquals(tracker.getScheduledMessages(100).size(), delayedMessagesInSnapshotValue); + assertEquals(tracker2.getScheduledMessages(100).size(), delayedMessagesInSnapshotValue); assertTrue(mockBucketSnapshotStorage.createExceptionQueue.isEmpty()); assertTrue(mockBucketSnapshotStorage.getMetaDataExceptionQueue.isEmpty()); assertTrue(mockBucketSnapshotStorage.getSegmentExceptionQueue.isEmpty()); assertTrue(mockBucketSnapshotStorage.deleteExceptionQueue.isEmpty()); + + tracker2.close(); } @Test(dataProvider = "delayedTracker") @@ -390,10 +421,13 @@ public void testMaxIndexesPerSegment(BucketDelayedDeliveryTracker tracker) { tracker.getImmutableBuckets().asMapOfRanges().forEach((k, bucket) -> { assertEquals(bucket.getLastSegmentEntryId(), 4); }); + + tracker.close(); } @Test(dataProvider = "delayedTracker") - public void testClear(BucketDelayedDeliveryTracker tracker) { + public void testClear(BucketDelayedDeliveryTracker tracker) + throws ExecutionException, InterruptedException, TimeoutException { for (int i = 1; i <= 1001; i++) { tracker.addMessage(i, i, i * 10); } @@ -402,11 +436,13 @@ public void testClear(BucketDelayedDeliveryTracker tracker) { assertTrue(tracker.getImmutableBuckets().asMapOfRanges().size() > 0); assertEquals(tracker.getLastMutableBucket().size(), 1); - tracker.clear(); + tracker.clear().get(1, TimeUnit.MINUTES); assertEquals(tracker.getNumberOfDelayedMessages(), 0); assertEquals(tracker.getImmutableBuckets().asMapOfRanges().size(), 0); assertEquals(tracker.getLastMutableBucket().size(), 0); assertEquals(tracker.getSharedBucketPriorityQueue().size(), 0); + + tracker.close(); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java index 292889e8c159a..09b7cbbf1b99d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java @@ -18,9 +18,15 @@ */ package org.apache.pulsar.broker.service.persistent; +import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.CURSOR_INTERNAL_PROPERTY_PREFIX; +import static org.testng.Assert.assertTrue; import java.util.List; +import java.util.Map; import java.util.concurrent.TimeUnit; import lombok.Cleanup; +import org.apache.bookkeeper.client.BKException; +import org.apache.bookkeeper.client.BookKeeper; +import org.apache.bookkeeper.client.LedgerHandle; import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory; @@ -103,4 +109,57 @@ public void testBucketDelayedDeliveryWithAllConsumersDisconnecting() throws Exce Awaitility.await().untilAsserted(() -> Assert.assertEquals(dispatcher2.getNumberOfDelayedMessages(), 1000)); Assert.assertEquals(bucketKeys, bucketKeys2); } + + + @Test + public void testUnsubscribe() throws Exception { + String topic = BrokerTestUtil.newUniqueName("persistent://public/default/testUnsubscribes"); + + @Cleanup + Consumer c1 = pulsarClient.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName("sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + + @Cleanup + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .create(); + + for (int i = 0; i < 1000; i++) { + producer.newMessage() + .value("msg") + .deliverAfter(1, TimeUnit.HOURS) + .send(); + } + + Dispatcher dispatcher = pulsar.getBrokerService().getTopicReference(topic).get().getSubscription("sub").getDispatcher(); + Awaitility.await().untilAsserted(() -> Assert.assertEquals(dispatcher.getNumberOfDelayedMessages(), 1000)); + + Map cursorProperties = + ((PersistentDispatcherMultipleConsumers) dispatcher).getCursor().getCursorProperties(); + List bucketIds = cursorProperties.entrySet().stream() + .filter(x -> x.getKey().startsWith(CURSOR_INTERNAL_PROPERTY_PREFIX + "delayed.bucket")).map( + x -> Long.valueOf(x.getValue())).toList(); + + assertTrue(bucketIds.size() > 0); + + c1.close(); + + restartBroker(); + + admin.topics().deleteSubscription(topic, "sub"); + + for (Long bucketId : bucketIds) { + try { + LedgerHandle ledgerHandle = + pulsarTestContext.getBookKeeperClient() + .openLedger(bucketId, BookKeeper.DigestType.CRC32C, new byte[]{}); + Assert.fail("Should fail"); + } catch (BKException.BKNoSuchLedgerExistsException e) { + // ignore it + } + } + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java index 13e67762ace6f..e6573ce88971d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java @@ -1204,24 +1204,7 @@ public void testManagedLedgerBookieClientStats() throws Exception { System.out.println(e.getKey() + ": " + e.getValue()) ); - List cm = (List) metrics.get(keyNameBySubstrings(metrics, - "pulsar_managedLedger_client", "bookkeeper_ml_scheduler_completed_tasks")); - assertEquals(cm.size(), 1); - assertEquals(cm.get(0).tags.get("cluster"), "test"); - - cm = (List) metrics.get( - keyNameBySubstrings(metrics, - "pulsar_managedLedger_client", "bookkeeper_ml_scheduler_queue")); - assertEquals(cm.size(), 1); - assertEquals(cm.get(0).tags.get("cluster"), "test"); - - cm = (List) metrics.get( - keyNameBySubstrings(metrics, - "pulsar_managedLedger_client", "bookkeeper_ml_scheduler_total_tasks")); - assertEquals(cm.size(), 1); - assertEquals(cm.get(0).tags.get("cluster"), "test"); - - cm = (List) metrics.get( + List cm = (List) metrics.get( keyNameBySubstrings(metrics, "pulsar_managedLedger_client", "bookkeeper_ml_scheduler_threads")); assertEquals(cm.size(), 1); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java index e8b9baa992c46..61c7a98602b69 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java @@ -22,16 +22,15 @@ import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; - import io.netty.channel.ChannelHandlerContext; - +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; - import lombok.Cleanup; +import org.apache.bookkeeper.common.util.JsonUtil; import org.apache.pulsar.client.impl.ConsumerBase; import org.apache.pulsar.client.impl.PartitionedProducerImpl; import org.apache.pulsar.client.impl.ProducerBase; @@ -831,4 +830,28 @@ public void testConsumerReconnect() throws Exception { mockBrokerService.resetHandleConnect(); mockBrokerService.resetHandleSubscribe(); } + + @Test + public void testCommandErrorMessageIsNull() throws Exception { + @Cleanup + PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getBrokerAddress()).build(); + + mockBrokerService.setHandleProducer((ctx, producer) -> { + try { + ctx.writeAndFlush(Commands.newError(producer.getRequestId(), ServerError.AuthorizationError, null)); + } catch (Exception e) { + fail("Send error command failed", e); + } + }); + + try { + client.newProducer().topic("persistent://prop/use/ns/t1").create(); + fail(); + } catch (Exception e) { + assertTrue(e instanceof PulsarClientException.AuthorizationException); + Map map = JsonUtil.fromJson(e.getMessage(), Map.class); + assertEquals(map.get("errorMsg"), ""); + } + mockBrokerService.resetHandleProducer(); + } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java index caa9cd3fd1daa..4a76170d116a3 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicPolicies.java @@ -20,6 +20,7 @@ import com.google.common.collect.Sets; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -188,4 +189,8 @@ public boolean isSchemaValidationEnforced() { public Set getReplicationClustersSet() { return replicationClusters != null ? Sets.newTreeSet(this.replicationClusters) : null; } + + public Map getSubscriptionPolicies() { + return subscriptionPolicies == null ? Collections.emptyMap() : subscriptionPolicies; + } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java index 8a5684cf676b0..85c4d021fdf22 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java @@ -368,7 +368,7 @@ public static BaseCommand newErrorCommand(long requestId, ServerError serverErro cmd.setError() .setRequestId(requestId) .setError(serverError) - .setMessage(message); + .setMessage(message != null ? message : ""); return cmd; } @@ -401,7 +401,7 @@ public static BaseCommand newSendErrorCommand(long producerId, long sequenceId, .setProducerId(producerId) .setSequenceId(sequenceId) .setError(error) - .setMessage(errorMsg); + .setMessage(errorMsg != null ? errorMsg : ""); return cmd; } diff --git a/pulsar-sql/presto-distribution/LICENSE b/pulsar-sql/presto-distribution/LICENSE index c523fae7606e1..0c73a9908182e 100644 --- a/pulsar-sql/presto-distribution/LICENSE +++ b/pulsar-sql/presto-distribution/LICENSE @@ -424,18 +424,21 @@ The Apache Software License, Version 2.0 - async-http-client-2.12.1.jar - async-http-client-netty-utils-2.12.1.jar * Apache Bookkeeper - - bookkeeper-common-4.15.4.jar - - bookkeeper-common-allocator-4.15.4.jar - - bookkeeper-proto-4.15.4.jar - - bookkeeper-server-4.15.4.jar - - bookkeeper-stats-api-4.15.4.jar - - bookkeeper-tools-framework-4.15.4.jar - - circe-checksum-4.15.4.jar - - codahale-metrics-provider-4.15.4.jar - - cpu-affinity-4.15.4.jar - - http-server-4.15.4.jar - - prometheus-metrics-provider-4.15.4.jar - - codahale-metrics-provider-4.15.4.jar + - bookkeeper-common-4.16.0.jar + - bookkeeper-common-allocator-4.16.0.jar + - bookkeeper-proto-4.16.0.jar + - bookkeeper-server-4.16.0.jar + - bookkeeper-stats-api-4.16.0.jar + - bookkeeper-tools-framework-4.16.0.jar + - circe-checksum-4.16.0.jar + - codahale-metrics-provider-4.16.0.jar + - cpu-affinity-4.16.0.jar + - http-server-4.16.0.jar + - prometheus-metrics-provider-4.16.0.jar + - codahale-metrics-provider-4.16.0.jar + - bookkeeper-slogger-api-4.16.0.jar + - bookkeeper-slogger-slf4j-4.16.0.jar + - native-io-4.16.0.jar * Apache Commons - commons-cli-1.5.0.jar - commons-codec-1.15.jar @@ -476,6 +479,8 @@ The Apache Software License, Version 2.0 - amqp-client-5.5.3.jar * Stream Lib - stream-2.9.5.jar + * High Performance Primitive Collections for Java + - hppc-0.9.1.jar Protocol Buffers License diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorMetricsTracker.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorMetricsTracker.java index b586063cc3c9a..12ee2da463c40 100644 --- a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorMetricsTracker.java +++ b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarConnectorMetricsTracker.java @@ -185,7 +185,7 @@ public void end_ENTRY_QUEUE_DEQUEUE_WAIT_TIME() { public void register_BYTES_READ(long bytes) { if (statsLogger != null) { bytesReadSum += bytes; - statsLoggerBytesRead.add(bytes); + statsLoggerBytesRead.addCount(bytes); } } @@ -220,7 +220,7 @@ public void end_MESSAGE_QUEUE_ENQUEUE_WAIT_TIME() { public void incr_NUM_MESSAGES_DESERIALIZED_PER_ENTRY() { if (statsLogger != null) { numMessagedDerserializedPerBatch++; - statsLoggerNumMessagesDeserialized.add(1); + statsLoggerNumMessagesDeserialized.addCount(1); } } @@ -295,7 +295,7 @@ public void end_RECORD_DESERIALIZE_TIME() { public void incr_NUM_RECORD_DESERIALIZED() { if (statsLogger != null) { - statsLoggerNumRecordDeserialized.add(1); + statsLoggerNumRecordDeserialized.addCount(1); } } diff --git a/testmocks/src/main/java/org/apache/bookkeeper/client/TestStatsProvider.java b/testmocks/src/main/java/org/apache/bookkeeper/client/TestStatsProvider.java index f015306cb048c..4d08a7f80df5b 100644 --- a/testmocks/src/main/java/org/apache/bookkeeper/client/TestStatsProvider.java +++ b/testmocks/src/main/java/org/apache/bookkeeper/client/TestStatsProvider.java @@ -59,10 +59,16 @@ public void dec() { } @Override - public void add(long delta) { + public void addCount(long delta) { updateMax(val.addAndGet(delta)); } + @Override + public void addLatency(long eventLatency, TimeUnit unit) { + long valueMillis = unit.toMillis(eventLatency); + updateMax(val.addAndGet(valueMillis)); + } + @Override public Long get() { return val.get();