You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zh...@apache.org on 2023/04/20 01:37:19 UTC
[pulsar] branch branch-3.0 updated (0df74125950 -> e1d63990644)
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a change to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from 0df74125950 [fix][fn] Supply download auth params when provided for k8s runtime (#20144)
new 1d1a3ef864a [improve] [broker] Close temporary open ledger in BookkeeperBucketSnapshotStorage (#20111)
new 49480ea558e [improve][broker] Make timer execute immediately after load index (#20126)
new ff59240165c [improve][broker] Optimize delayed metadata index bitmap (#20136)
new e1d63990644 [improve][broker] Cache LedgerHandle in BookkeeperBucketSnapshotStorage (#20117)
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../bucket/BookkeeperBucketSnapshotStorage.java | 33 +++++++++++---
.../bucket/BucketDelayedDeliveryTracker.java | 11 +++--
.../broker/delayed/bucket/ImmutableBucket.java | 53 +++++++++++++---------
.../broker/delayed/bucket/MutableBucket.java | 11 +++--
.../BookkeeperBucketSnapshotStorageTest.java | 43 ++++++++++++++++++
5 files changed, 115 insertions(+), 36 deletions(-)
[pulsar] 04/04: [improve][broker] Cache LedgerHandle in BookkeeperBucketSnapshotStorage (#20117)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e1d63990644700bf61b3d7af1ef6d4d62145c2bb
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Thu Apr 20 09:26:30 2023 +0800
[improve][broker] Cache LedgerHandle in BookkeeperBucketSnapshotStorage (#20117)
(cherry picked from commit d3fa998aa7c0a7a9452079ef2ff05bccf6b273cf)
---
.../bucket/BookkeeperBucketSnapshotStorage.java | 52 +++++++++++-----------
.../BookkeeperBucketSnapshotStorageTest.java | 43 ++++++++++++++++++
2 files changed, 69 insertions(+), 26 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
index 9c30ccf1c0b..18a4c322f7b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import javax.validation.constraints.NotNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BKException;
@@ -48,6 +49,8 @@ public class BookkeeperBucketSnapshotStorage implements BucketSnapshotStorage {
private final ServiceConfiguration config;
private BookKeeper bookKeeper;
+ private final Map<Long, CompletableFuture<LedgerHandle>> ledgerHandleFutureCache = new ConcurrentHashMap<>();
+
public BookkeeperBucketSnapshotStorage(PulsarService pulsar) {
this.pulsar = pulsar;
this.config = pulsar.getConfig();
@@ -66,45 +69,30 @@ public class BookkeeperBucketSnapshotStorage implements BucketSnapshotStorage {
@Override
public CompletableFuture<SnapshotMetadata> getBucketSnapshotMetadata(long bucketId) {
- return openLedger(bucketId).thenCompose(ledgerHandle -> {
- CompletableFuture<SnapshotMetadata> snapshotFuture =
- getLedgerEntry(ledgerHandle, 0, 0)
- .thenApply(entryEnumeration -> parseSnapshotMetadataEntry(entryEnumeration.nextElement()));
-
- snapshotFuture.whenComplete((__, e) -> closeLedger(ledgerHandle));
-
- return snapshotFuture;
- });
+ return getLedgerHandle(bucketId).thenCompose(ledgerHandle -> getLedgerEntry(ledgerHandle, 0, 0)
+ .thenApply(entryEnumeration -> parseSnapshotMetadataEntry(entryEnumeration.nextElement())));
}
@Override
public CompletableFuture<List<SnapshotSegment>> getBucketSnapshotSegment(long bucketId, long firstSegmentEntryId,
long lastSegmentEntryId) {
- return openLedger(bucketId).thenCompose(ledgerHandle -> {
- CompletableFuture<List<SnapshotSegment>> parseFuture =
- getLedgerEntry(ledgerHandle, firstSegmentEntryId, lastSegmentEntryId)
- .thenApply(this::parseSnapshotSegmentEntries);
-
- parseFuture.whenComplete((__, e) -> closeLedger(ledgerHandle));
-
- return parseFuture;
- });
+ return getLedgerHandle(bucketId).thenCompose(
+ ledgerHandle -> getLedgerEntry(ledgerHandle, firstSegmentEntryId, lastSegmentEntryId)
+ .thenApply(this::parseSnapshotSegmentEntries));
}
@Override
public CompletableFuture<Long> getBucketSnapshotLength(long bucketId) {
- return openLedger(bucketId).thenCompose(ledgerHandle -> {
- CompletableFuture<Long> lengthFuture =
- CompletableFuture.completedFuture(ledgerHandle.getLength());
-
- lengthFuture.whenComplete((__, e) -> closeLedger(ledgerHandle));
-
- return lengthFuture;
- });
+ return getLedgerHandle(bucketId).thenCompose(
+ ledgerHandle -> CompletableFuture.completedFuture(ledgerHandle.getLength()));
}
@Override
public CompletableFuture<Void> deleteBucketSnapshot(long bucketId) {
+ CompletableFuture<LedgerHandle> ledgerHandleFuture = ledgerHandleFutureCache.remove(bucketId);
+ if (ledgerHandleFuture != null) {
+ ledgerHandleFuture.whenComplete((lh, ex) -> closeLedger(lh));
+ }
return deleteLedger(bucketId);
}
@@ -178,6 +166,18 @@ public class BookkeeperBucketSnapshotStorage implements BucketSnapshotStorage {
return future;
}
+ private CompletableFuture<LedgerHandle> getLedgerHandle(Long ledgerId) {
+ CompletableFuture<LedgerHandle> ledgerHandleCompletableFuture =
+ ledgerHandleFutureCache.computeIfAbsent(ledgerId, k -> openLedger(ledgerId));
+ // remove future of completed exceptionally
+ ledgerHandleCompletableFuture.whenComplete((__, ex) -> {
+ if (ex != null) {
+ ledgerHandleFutureCache.remove(ledgerId, ledgerHandleCompletableFuture);
+ }
+ });
+ return ledgerHandleCompletableFuture;
+ }
+
private CompletableFuture<LedgerHandle> openLedger(Long ledgerId) {
final CompletableFuture<LedgerHandle> future = new CompletableFuture<>();
bookKeeper.asyncOpenLedger(
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 a628b58e10d..7cb6b8d5865 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
@@ -30,6 +30,7 @@ import java.util.concurrent.ExecutionException;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.delayed.bucket.BookkeeperBucketSnapshotStorage;
import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat;
+import org.apache.pulsar.common.util.FutureUtil;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -204,4 +205,46 @@ public class BookkeeperBucketSnapshotStorageTest extends MockedPulsarServiceBase
Assert.assertTrue(bucketSnapshotLength > 0L);
}
+ @Test
+ public void testConcurrencyGet() throws ExecutionException, InterruptedException {
+ DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata segmentMetadata =
+ DelayedMessageIndexBucketSnapshotFormat.SnapshotSegmentMetadata.newBuilder()
+ .setMinScheduleTimestamp(System.currentTimeMillis())
+ .setMaxScheduleTimestamp(System.currentTimeMillis())
+ .putDelayedIndexBitMap(100L, ByteString.copyFrom(new byte[1])).build();
+
+ DelayedMessageIndexBucketSnapshotFormat.SnapshotMetadata snapshotMetadata =
+ DelayedMessageIndexBucketSnapshotFormat.SnapshotMetadata.newBuilder()
+ .addMetadataList(segmentMetadata)
+ .build();
+ List<DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment> bucketSnapshotSegments = new ArrayList<>();
+
+ long timeMillis = System.currentTimeMillis();
+ DelayedMessageIndexBucketSnapshotFormat.DelayedIndex delayedIndex =
+ DelayedMessageIndexBucketSnapshotFormat.DelayedIndex.newBuilder().setLedgerId(100L).setEntryId(10L)
+ .setTimestamp(timeMillis).build();
+ DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment snapshotSegment =
+ DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment.newBuilder().addIndexes(delayedIndex).build();
+ bucketSnapshotSegments.add(snapshotSegment);
+ bucketSnapshotSegments.add(snapshotSegment);
+
+ CompletableFuture<Long> future =
+ bucketSnapshotStorage.createBucketSnapshot(snapshotMetadata,
+ bucketSnapshotSegments, UUID.randomUUID().toString(), TOPIC_NAME, CURSOR_NAME);
+ Long bucketId = future.get();
+ Assert.assertNotNull(bucketId);
+
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ CompletableFuture<Void> future0 = CompletableFuture.runAsync(() -> {
+ List<DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment> list =
+ bucketSnapshotStorage.getBucketSnapshotSegment(bucketId, 1, 3).join();
+ Assert.assertTrue(list.size() > 0);
+ });
+ futures.add(future0);
+ }
+
+ FutureUtil.waitForAll(futures).join();
+ }
+
}
[pulsar] 03/04: [improve][broker] Optimize delayed metadata index bitmap (#20136)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ff59240165c73a9c3a3dcca20702ab44b0b18d33
Author: Qiang Zhao <ma...@apache.org>
AuthorDate: Thu Apr 20 08:21:08 2023 +0800
[improve][broker] Optimize delayed metadata index bitmap (#20136)
(cherry picked from commit 6dc0b0ea38ab9dc410a24b19fd7567b9e013837d)
---
.../bucket/BucketDelayedDeliveryTracker.java | 3 ++
.../broker/delayed/bucket/ImmutableBucket.java | 53 +++++++++++++---------
.../broker/delayed/bucket/MutableBucket.java | 11 +++--
3 files changed, 42 insertions(+), 25 deletions(-)
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 ad2fc6fae4c..5d1dc2e2703 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
@@ -485,6 +485,9 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
});
});
}
+
+ // optimize bm
+ delayedIndexBitMap.values().forEach(RoaringBitmap::runOptimize);
immutableBucketDelayedIndexPair.getLeft().setDelayedIndexBitMap(delayedIndexBitMap);
afterCreateImmutableBucket(immutableBucketDelayedIndexPair, createStartTime);
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 82e98cefa5d..57de5c84fcd 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,11 +20,10 @@ 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.NULL_LONG_PROMISE;
-import com.google.protobuf.ByteString;
+import java.io.IOException;
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.function.Supplier;
@@ -37,8 +36,8 @@ import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotF
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.InvalidRoaringFormat;
import org.roaringbitmap.RoaringBitmap;
-import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
@Slf4j
class ImmutableBucket extends Bucket {
@@ -98,7 +97,7 @@ class ImmutableBucket extends Bucket {
this.setLastSegmentEntryId(metadataList.size());
this.recoverDelayedIndexBitMapAndNumber(nextSnapshotEntryIndex, metadataList);
List<Long> firstScheduleTimestamps = metadataList.stream().map(
- SnapshotSegmentMetadata::getMinScheduleTimestamp).toList();
+ SnapshotSegmentMetadata::getMinScheduleTimestamp).toList();
this.setFirstScheduleTimestamps(firstScheduleTimestamps);
return nextSnapshotEntryIndex + 1;
@@ -139,25 +138,37 @@ class ImmutableBucket extends Bucket {
});
}
+ /**
+ * Recover delayed index bit map and message numbers.
+ * @throws InvalidRoaringFormat invalid bitmap serialization format
+ */
private void recoverDelayedIndexBitMapAndNumber(int startSnapshotIndex,
- List<SnapshotSegmentMetadata> segmentMetadata) {
- this.delayedIndexBitMap.clear();
- MutableLong numberMessages = new MutableLong(0);
- for (int i = startSnapshotIndex; i < segmentMetadata.size(); i++) {
- Map<Long, ByteString> bitByteStringMap = segmentMetadata.get(i).getDelayedIndexBitMapMap();
- bitByteStringMap.forEach((leaderId, bitSetString) -> {
- boolean exist = this.delayedIndexBitMap.containsKey(leaderId);
- RoaringBitmap bitSet =
- new ImmutableRoaringBitmap(bitSetString.asReadOnlyByteBuffer()).toRoaringBitmap();
- numberMessages.add(bitSet.getCardinality());
- if (!exist) {
- this.delayedIndexBitMap.put(leaderId, bitSet);
- } else {
- this.delayedIndexBitMap.get(leaderId).or(bitSet);
+ List<SnapshotSegmentMetadata> segmentMetaList) {
+ delayedIndexBitMap.clear(); // cleanup dirty bm
+ final var numberMessages = new MutableLong(0);
+ for (int i = startSnapshotIndex; i < segmentMetaList.size(); i++) {
+ for (final var entry : segmentMetaList.get(i).getDelayedIndexBitMapMap().entrySet()) {
+ final var ledgerId = entry.getKey();
+ final var bs = entry.getValue();
+ final var sbm = new RoaringBitmap();
+ try {
+ sbm.deserialize(bs.asReadOnlyByteBuffer());
+ } catch (IOException e) {
+ throw new InvalidRoaringFormat(e.getMessage());
}
- });
+ numberMessages.add(sbm.getCardinality());
+ delayedIndexBitMap.compute(ledgerId, (lId, bm) -> {
+ if (bm == null) {
+ return sbm;
+ }
+ bm.or(sbm);
+ return bm;
+ });
+ }
}
- this.setNumberBucketDelayedMessages(numberMessages.getValue());
+ // optimize bm
+ delayedIndexBitMap.values().forEach(RoaringBitmap::runOptimize);
+ setNumberBucketDelayedMessages(numberMessages.getValue());
}
CompletableFuture<List<DelayedMessageIndexBucketSnapshotFormat.SnapshotSegment>> getRemainSnapshotSegment() {
@@ -193,7 +204,7 @@ class ImmutableBucket extends Bucket {
stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.delete);
} else {
log.info("[{}] Delete bucket snapshot finish, bucketId: {}, bucketKey: {}",
- dispatcherName, bucketId, bucketKey);
+ dispatcherName, bucketId, bucketKey);
stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.delete,
System.currentTimeMillis() - deleteStartTime);
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 e49ebe9606e..f404d5d02c1 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
@@ -116,10 +116,13 @@ class MutableBucket extends Bucket implements AutoCloseable {
Iterator<Map.Entry<Long, RoaringBitmap>> iterator = bitMap.entrySet().iterator();
while (iterator.hasNext()) {
- Map.Entry<Long, RoaringBitmap> entry = iterator.next();
- byte[] array = new byte[entry.getValue().serializedSizeInBytes()];
- entry.getValue().serialize(ByteBuffer.wrap(array));
- segmentMetadataBuilder.putDelayedIndexBitMap(entry.getKey(), ByteString.copyFrom(array));
+ final var entry = iterator.next();
+ final var lId = entry.getKey();
+ final var bm = entry.getValue();
+ bm.runOptimize();
+ final var array = new byte[bm.serializedSizeInBytes()];
+ bm.serialize(ByteBuffer.wrap(array));
+ segmentMetadataBuilder.putDelayedIndexBitMap(lId, ByteString.copyFrom(array));
iterator.remove();
}
[pulsar] 01/04: [improve] [broker] Close temporary open ledger in BookkeeperBucketSnapshotStorage (#20111)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1d1a3ef864a65c995ceda4b7875ed934c2574298
Author: lifepuzzlefun <wj...@163.com>
AuthorDate: Mon Apr 17 19:32:13 2023 +0800
[improve] [broker] Close temporary open ledger in BookkeeperBucketSnapshotStorage (#20111)
(cherry picked from commit b50e8802a5224dd68832e263e7046650771a1a4e)
---
.../bucket/BookkeeperBucketSnapshotStorage.java | 33 +++++++++++++++++-----
1 file changed, 26 insertions(+), 7 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
index e7d4f9301dd..9c30ccf1c0b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java
@@ -66,22 +66,41 @@ public class BookkeeperBucketSnapshotStorage implements BucketSnapshotStorage {
@Override
public CompletableFuture<SnapshotMetadata> getBucketSnapshotMetadata(long bucketId) {
- return openLedger(bucketId).thenCompose(
- ledgerHandle -> getLedgerEntry(ledgerHandle, 0, 0).
- thenApply(entryEnumeration -> parseSnapshotMetadataEntry(entryEnumeration.nextElement())));
+ return openLedger(bucketId).thenCompose(ledgerHandle -> {
+ CompletableFuture<SnapshotMetadata> snapshotFuture =
+ getLedgerEntry(ledgerHandle, 0, 0)
+ .thenApply(entryEnumeration -> parseSnapshotMetadataEntry(entryEnumeration.nextElement()));
+
+ snapshotFuture.whenComplete((__, e) -> closeLedger(ledgerHandle));
+
+ return snapshotFuture;
+ });
}
@Override
public CompletableFuture<List<SnapshotSegment>> getBucketSnapshotSegment(long bucketId, long firstSegmentEntryId,
long lastSegmentEntryId) {
- return openLedger(bucketId).thenCompose(
- ledgerHandle -> getLedgerEntry(ledgerHandle, firstSegmentEntryId,
- lastSegmentEntryId).thenApply(this::parseSnapshotSegmentEntries));
+ return openLedger(bucketId).thenCompose(ledgerHandle -> {
+ CompletableFuture<List<SnapshotSegment>> parseFuture =
+ getLedgerEntry(ledgerHandle, firstSegmentEntryId, lastSegmentEntryId)
+ .thenApply(this::parseSnapshotSegmentEntries);
+
+ parseFuture.whenComplete((__, e) -> closeLedger(ledgerHandle));
+
+ return parseFuture;
+ });
}
@Override
public CompletableFuture<Long> getBucketSnapshotLength(long bucketId) {
- return openLedger(bucketId).thenApply(LedgerHandle::getLength);
+ return openLedger(bucketId).thenCompose(ledgerHandle -> {
+ CompletableFuture<Long> lengthFuture =
+ CompletableFuture.completedFuture(ledgerHandle.getLength());
+
+ lengthFuture.whenComplete((__, e) -> closeLedger(ledgerHandle));
+
+ return lengthFuture;
+ });
}
@Override
[pulsar] 02/04: [improve][broker] Make timer execute immediately after load index (#20126)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhaocong pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 49480ea558e647169e8df01bfd2e871a5386e19e
Author: Cong Zhao <zh...@apache.org>
AuthorDate: Wed Apr 19 16:44:12 2023 +0800
[improve][broker] Make timer execute immediately after load index (#20126)
(cherry picked from commit 9b723022436cc1a150af765103e8d343679f92ce)
---
.../broker/delayed/bucket/BucketDelayedDeliveryTracker.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
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 6678c6df254..ad2fc6fae4c 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
@@ -541,7 +541,7 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
@Override
public synchronized NavigableSet<PositionImpl> getScheduledMessages(int maxMessages) {
- if (!checkPendingOpDone()) {
+ if (!checkPendingLoadDone()) {
if (log.isDebugEnabled()) {
log.debug("[{}] Skip getScheduledMessages to wait for bucket snapshot load finish.",
dispatcher.getName());
@@ -628,11 +628,11 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
if (timeout != null) {
timeout.cancel();
}
- timeout = timer.newTimeout(this, tickTimeMillis, TimeUnit.MILLISECONDS);
+ timeout = timer.newTimeout(this, 0, TimeUnit.MILLISECONDS);
}
});
- if (!checkPendingOpDone() || loadFuture.isCompletedExceptionally()) {
+ if (!checkPendingLoadDone() || loadFuture.isCompletedExceptionally()) {
break;
}
}
@@ -651,7 +651,7 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker
return positions;
}
- private synchronized boolean checkPendingOpDone() {
+ private synchronized boolean checkPendingLoadDone() {
if (pendingLoad == null || pendingLoad.isDone()) {
pendingLoad = null;
return true;