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:20 UTC
[pulsar] 01/04: [improve] [broker] Close temporary open ledger in BookkeeperBucketSnapshotStorage (#20111)
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