You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/01/03 12:17:55 UTC
(pulsar) 02/02: [improve][ml] Optimize BlobStoreManagedLedgerOffloader.getOffloadPolicies (#23776)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit d55e245e5f32c2630d6944bf160153e7f2abdf3a
Author: 道君 <da...@apache.org>
AuthorDate: Wed Dec 25 14:37:15 2024 +0800
[improve][ml] Optimize BlobStoreManagedLedgerOffloader.getOffloadPolicies (#23776)
(cherry picked from commit 369c352664a68acbb9bb3584b433498440a41034)
---
.../offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java | 12 +++++++-----
1 file changed, 7 insertions(+), 5 deletions(-)
diff --git a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java
index b4ed940c9cd..d6d298159ea 100644
--- a/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java
+++ b/tiered-storage/jcloud/src/main/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java
@@ -98,6 +98,7 @@ public class BlobStoreManagedLedgerOffloader implements LedgerOffloader {
private final OrderedScheduler scheduler;
private final TieredStorageConfiguration config;
+ private final OffloadPolicies policies;
private final Location writeLocation;
// metadata to be stored as part of the offloaded ledger metadata
@@ -105,8 +106,8 @@ public class BlobStoreManagedLedgerOffloader implements LedgerOffloader {
private final ConcurrentMap<BlobStoreLocation, BlobStore> blobStores = new ConcurrentHashMap<>();
private OffloadSegmentInfoImpl segmentInfo;
- private AtomicLong bufferLength = new AtomicLong(0);
- private AtomicLong segmentLength = new AtomicLong(0);
+ private final AtomicLong bufferLength = new AtomicLong(0);
+ private final AtomicLong segmentLength = new AtomicLong(0);
private final long maxBufferLength;
private final OffsetsCache entryOffsetsCache;
private final ConcurrentLinkedQueue<Entry> offloadBuffer = new ConcurrentLinkedQueue<>();
@@ -138,6 +139,9 @@ public class BlobStoreManagedLedgerOffloader implements LedgerOffloader {
this.scheduler = scheduler;
this.userMetadata = userMetadata;
this.config = config;
+ Properties properties = new Properties();
+ properties.putAll(config.getConfigProperties());
+ this.policies = OffloadPoliciesImpl.create(properties);
this.streamingBlockSize = config.getMinBlockSizeInBytes();
this.maxSegmentCloseTime = Duration.ofSeconds(config.getMaxSegmentTimeInSecond());
this.maxSegmentLength = config.getMaxSegmentSizeInBytes();
@@ -658,9 +662,7 @@ public class BlobStoreManagedLedgerOffloader implements LedgerOffloader {
@Override
public OffloadPolicies getOffloadPolicies() {
- Properties properties = new Properties();
- properties.putAll(config.getConfigProperties());
- return OffloadPoliciesImpl.create(properties);
+ return this.policies;
}
@Override