You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/10/11 01:56:14 UTC
[GitHub] [pulsar] zymap commented on a diff in pull request #17398: [feat][tiered-storage] ADD offload_time_threshold policy
zymap commented on code in PR #17398:
URL: https://github.com/apache/pulsar/pull/17398#discussion_r991212702
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -2354,72 +2355,105 @@ private void scheduleDeferredTrimming(boolean isTruncate, CompletableFuture<?> p
}
private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
- if (config.getLedgerOffloader() != null
- && config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
- && config.getLedgerOffloader().getOffloadPolicies() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0) {
+ if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
+ || config.getLedgerOffloader().getOffloadPolicies() == null) {
+ return;
+ }
+
+ final OffloadPoliciesImpl policies = config.getLedgerOffloader().getOffloadPolicies();
+ final long offloadThresholdInBytes =
+ Optional.ofNullable(policies.getManagedLedgerOffloadThresholdInBytes()).orElse(-1L);
+ final long offloadThresholdInSeconds =
+ Optional.ofNullable(policies.getManagedLedgerOffloadThresholdInSeconds()).orElse(-1L);
+ if (offloadThresholdInBytes >= 0 || offloadThresholdInSeconds >= 0) {
executor.executeOrdered(name, safeRun(() -> maybeOffload(promise)));
}
}
private void maybeOffload(CompletableFuture<PositionImpl> finalPromise) {
if (!offloadMutex.tryLock()) {
scheduledExecutor.schedule(safeRun(() -> maybeOffloadInBackground(finalPromise)),
- 100, TimeUnit.MILLISECONDS);
- } else {
- CompletableFuture<PositionImpl> unlockingPromise = new CompletableFuture<>();
- unlockingPromise.whenComplete((res, ex) -> {
- offloadMutex.unlock();
- if (ex != null) {
- finalPromise.completeExceptionally(ex);
- } else {
- finalPromise.complete(res);
- }
- });
+ 100, TimeUnit.MILLISECONDS);
+ return;
+ }
- if (config.getLedgerOffloader() != null
- && config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
- && config.getLedgerOffloader().getOffloadPolicies() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes()
- != null) {
- long threshold = config.getLedgerOffloader().getOffloadPolicies()
- .getManagedLedgerOffloadThresholdInBytes();
-
- long sizeSummed = 0;
- long alreadyOffloadedSize = 0;
- long toOffloadSize = 0;
-
- ConcurrentLinkedDeque<LedgerInfo> toOffload = new ConcurrentLinkedDeque<>();
-
- // go through ledger list from newest to oldest and build a list to offload in oldest to newest order
- for (Map.Entry<Long, LedgerInfo> e : ledgers.descendingMap().entrySet()) {
- long size = e.getValue().getSize();
- sizeSummed += size;
- boolean alreadyOffloaded = e.getValue().hasOffloadContext()
- && e.getValue().getOffloadContext().getComplete();
- if (alreadyOffloaded) {
- alreadyOffloadedSize += size;
- } else if (sizeSummed > threshold) {
- toOffloadSize += size;
- toOffload.addFirst(e.getValue());
- }
- }
+ CompletableFuture<PositionImpl> unlockingPromise = new CompletableFuture<>();
+ unlockingPromise.whenComplete((res, ex) -> {
+ offloadMutex.unlock();
+ if (ex != null) {
+ finalPromise.completeExceptionally(ex);
+ } else {
+ finalPromise.complete(res);
+ }
+ });
- if (toOffload.size() > 0) {
- log.info("[{}] Going to automatically offload ledgers {}"
- + ", total size = {}, already offloaded = {}, to offload = {}",
- name, toOffload.stream().map(LedgerInfo::getLedgerId).collect(Collectors.toList()),
- sizeSummed, alreadyOffloadedSize, toOffloadSize);
- offloadLoop(unlockingPromise, toOffload, PositionImpl.LATEST, Optional.empty());
- } else {
- // offloadLoop will complete immediately with an empty list to offload
- log.debug("[{}] Nothing to offload, total size = {}, already offloaded = {}, threshold = {}",
- name, sizeSummed, alreadyOffloadedSize, threshold);
- unlockingPromise.complete(PositionImpl.LATEST);
+ if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
+ || config.getLedgerOffloader().getOffloadPolicies() == null) {
+ log.debug("[{}] Nothing to offload due to offloader or offloadPolicies is NULL", name);
+ finalPromise.complete(PositionImpl.LATEST);
Review Comment:
When you trigger the offload, no data offloaded successfully. But there is no error in the broker, that doesn't make sense.
I think this case won't happen. But we also shouldn't make it complete successfully. I think it's better to make the promise fail.
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java:
##########
@@ -839,7 +887,9 @@ public CompletableFuture<Void> offload(ReadHandle ledger,
for (int i = 0; i < 25; i++) {
ledger.addEntry(buildEntry(10, "entry-" + i));
}
+ System.out.println("before");
offloadRunning.await();
+ System.out.println("after");
Review Comment:
Remove unused output?
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java:
##########
@@ -839,7 +887,9 @@ public CompletableFuture<Void> offload(ReadHandle ledger,
for (int i = 0; i < 25; i++) {
ledger.addEntry(buildEntry(10, "entry-" + i));
}
+ System.out.println("before");
Review Comment:
Remove unused output?
##########
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java:
##########
@@ -2354,72 +2355,105 @@ private void scheduleDeferredTrimming(boolean isTruncate, CompletableFuture<?> p
}
private void maybeOffloadInBackground(CompletableFuture<PositionImpl> promise) {
- if (config.getLedgerOffloader() != null
- && config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
- && config.getLedgerOffloader().getOffloadPolicies() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes() >= 0) {
+ if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
+ || config.getLedgerOffloader().getOffloadPolicies() == null) {
+ return;
+ }
+
+ final OffloadPoliciesImpl policies = config.getLedgerOffloader().getOffloadPolicies();
+ final long offloadThresholdInBytes =
+ Optional.ofNullable(policies.getManagedLedgerOffloadThresholdInBytes()).orElse(-1L);
+ final long offloadThresholdInSeconds =
+ Optional.ofNullable(policies.getManagedLedgerOffloadThresholdInSeconds()).orElse(-1L);
+ if (offloadThresholdInBytes >= 0 || offloadThresholdInSeconds >= 0) {
executor.executeOrdered(name, safeRun(() -> maybeOffload(promise)));
}
}
private void maybeOffload(CompletableFuture<PositionImpl> finalPromise) {
if (!offloadMutex.tryLock()) {
scheduledExecutor.schedule(safeRun(() -> maybeOffloadInBackground(finalPromise)),
- 100, TimeUnit.MILLISECONDS);
- } else {
- CompletableFuture<PositionImpl> unlockingPromise = new CompletableFuture<>();
- unlockingPromise.whenComplete((res, ex) -> {
- offloadMutex.unlock();
- if (ex != null) {
- finalPromise.completeExceptionally(ex);
- } else {
- finalPromise.complete(res);
- }
- });
+ 100, TimeUnit.MILLISECONDS);
+ return;
+ }
- if (config.getLedgerOffloader() != null
- && config.getLedgerOffloader() != NullLedgerOffloader.INSTANCE
- && config.getLedgerOffloader().getOffloadPolicies() != null
- && config.getLedgerOffloader().getOffloadPolicies().getManagedLedgerOffloadThresholdInBytes()
- != null) {
- long threshold = config.getLedgerOffloader().getOffloadPolicies()
- .getManagedLedgerOffloadThresholdInBytes();
-
- long sizeSummed = 0;
- long alreadyOffloadedSize = 0;
- long toOffloadSize = 0;
-
- ConcurrentLinkedDeque<LedgerInfo> toOffload = new ConcurrentLinkedDeque<>();
-
- // go through ledger list from newest to oldest and build a list to offload in oldest to newest order
- for (Map.Entry<Long, LedgerInfo> e : ledgers.descendingMap().entrySet()) {
- long size = e.getValue().getSize();
- sizeSummed += size;
- boolean alreadyOffloaded = e.getValue().hasOffloadContext()
- && e.getValue().getOffloadContext().getComplete();
- if (alreadyOffloaded) {
- alreadyOffloadedSize += size;
- } else if (sizeSummed > threshold) {
- toOffloadSize += size;
- toOffload.addFirst(e.getValue());
- }
- }
+ CompletableFuture<PositionImpl> unlockingPromise = new CompletableFuture<>();
+ unlockingPromise.whenComplete((res, ex) -> {
+ offloadMutex.unlock();
+ if (ex != null) {
+ finalPromise.completeExceptionally(ex);
+ } else {
+ finalPromise.complete(res);
+ }
+ });
- if (toOffload.size() > 0) {
- log.info("[{}] Going to automatically offload ledgers {}"
- + ", total size = {}, already offloaded = {}, to offload = {}",
- name, toOffload.stream().map(LedgerInfo::getLedgerId).collect(Collectors.toList()),
- sizeSummed, alreadyOffloadedSize, toOffloadSize);
- offloadLoop(unlockingPromise, toOffload, PositionImpl.LATEST, Optional.empty());
- } else {
- // offloadLoop will complete immediately with an empty list to offload
- log.debug("[{}] Nothing to offload, total size = {}, already offloaded = {}, threshold = {}",
- name, sizeSummed, alreadyOffloadedSize, threshold);
- unlockingPromise.complete(PositionImpl.LATEST);
+ if (config.getLedgerOffloader() == null || config.getLedgerOffloader() == NullLedgerOffloader.INSTANCE
+ || config.getLedgerOffloader().getOffloadPolicies() == null) {
+ log.debug("[{}] Nothing to offload due to offloader or offloadPolicies is NULL", name);
+ finalPromise.complete(PositionImpl.LATEST);
+ return;
+ }
+
+ long sizeSummed = 0;
+ long toOffloadSize = 0;
+ long alreadyOffloadedSize = 0;
+ ConcurrentLinkedDeque<LedgerInfo> toOffload = new ConcurrentLinkedDeque<>();
+
+ final OffloadPoliciesImpl policies = config.getLedgerOffloader().getOffloadPolicies();
+ final long offloadThresholdInBytes =
+ Optional.ofNullable(policies.getManagedLedgerOffloadThresholdInBytes()).orElse(-1L);
+ final long offloadTimeThresholdMillis =
+ Optional.ofNullable(policies.getManagedLedgerOffloadThresholdInSeconds()).filter(v -> v >= 0)
+ .map(TimeUnit.SECONDS::toMillis).orElse(-1L);
+
+ //Skip the following steps if `offloadTimeThreshold` and `offloadThresholdInBytes` are null OR negative values.
+ if (offloadThresholdInBytes < 0 && offloadTimeThresholdMillis < 0) {
+ log.debug("[{}] Nothing to offload due to [managedLedgerOffloadAutoTriggerSizeThresholdBytes] "
+ + "and [managedLedgerOffloadThresholdInSeconds] are null OR negative values.", name);
+ unlockingPromise.complete(PositionImpl.LATEST);
+ return;
+ }
+
+ for (Map.Entry<Long, LedgerInfo> e : ledgers.descendingMap().entrySet()) {
+ final LedgerInfo info = e.getValue();
+ // Skip current active ledger, an active ledger can't be offloaded.
+ // Can't `info.getLedgerId() == currentLedger.getId()` here, trigger offloading is before create ledger.
+ if (info.getTimestamp() == 0L) {
+ continue;
+ }
+
+ final long size = info.getSize();
+ final long timestamp = info.getTimestamp();
+ final long now = System.currentTimeMillis();
+ sizeSummed += size;
+
+ final boolean alreadyOffloaded = info.hasOffloadContext() && info.getOffloadContext().getComplete();
+ if (alreadyOffloaded) {
+ alreadyOffloadedSize += size;
+ } else {
+ if ((offloadThresholdInBytes >= 0 && sizeSummed > offloadThresholdInBytes)
Review Comment:
Why it makes the ledger offload earlier? Doesn't the `offloadThresholdInBytes` mean how much data I want to keep in the bookkeeper?
##########
managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/OffloadPrefixTest.java:
##########
@@ -787,33 +790,77 @@ private static byte[] buildEntry(int size, String pattern) {
return entry;
}
- @Test
- public void testAutoTriggerOffload() throws Exception {
+ @DataProvider(name = "testAutoTriggerOffload")
+ public Object[][] testAutoTriggerOffloadProvider() {
+ return new Object[][]{
+ {null, 0L},
+ {100L, null},
+ {-1L, null},
+ {null, null},
+ {-1L, -1L},
+ {1L, 1L}
+ };
+ }
+
+ @Test(dataProvider = "testAutoTriggerOffload")
+ public void testAutoTriggerOffload(Long sizeThreshold, Long timeThreshold) throws Exception {
MockLedgerOffloader offloader = new MockLedgerOffloader();
ManagedLedgerConfig config = new ManagedLedgerConfig();
config.setMaxEntriesPerLedger(10);
config.setRetentionTime(10, TimeUnit.MINUTES);
config.setRetentionSizeInMB(10);
- offloader.getOffloadPolicies().setManagedLedgerOffloadThresholdInBytes(100L);
+ offloader.getOffloadPolicies().setManagedLedgerOffloadThresholdInBytes(sizeThreshold);
+ offloader.getOffloadPolicies().setManagedLedgerOffloadThresholdInSeconds(timeThreshold);
config.setLedgerOffloader(offloader);
- ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("my_test_ledger", config);
+ ManagedLedgerImpl ledger = (ManagedLedgerImpl)factory.open("my_test_ledger" + UUID.randomUUID(), config);
// Ledger will roll twice, offload will run on first ledger after second closed
for (int i = 0; i < 25; i++) {
+ Thread.sleep(10);
ledger.addEntry(buildEntry(10, "entry-" + i));
}
assertEquals(ledger.getLedgersInfoAsList().size(), 3);
- // offload should eventually be triggered
- assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 1);
- assertEquals(offloader.offloadedLedgers(),
- Set.of(ledger.getLedgersInfoAsList().get(0).getLedgerId()));
+ if (sizeThreshold == null && timeThreshold != null && timeThreshold == 0L) {
+ // All the inactive ledgers will be offloaded
+ assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 2);
+ List<Long> allLedgerIds = ledger.getLedgersInfoAsList().stream().map(LedgerInfo::getLedgerId).toList();
+ assertEquals(offloader.offloadedLedgers(), Set.of(allLedgerIds.get(0), allLedgerIds.get(1)));
+ } else if (sizeThreshold != null && sizeThreshold == 100L && timeThreshold == null) {
+ // The last 2 ledgers won't be offloaded
+ assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 1);
+ List<Long> allLedgerIds = ledger.getLedgersInfoAsList().stream().map(LedgerInfo::getLedgerId).toList();
+ assertEquals(offloader.offloadedLedgers(), Set.of(allLedgerIds.get(0)));
+ } else if (sizeThreshold != null && sizeThreshold == -1L && timeThreshold == null) {
+ // Offloading is disabled, no ledgers will be offloaded.
+ assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 0);
+ assertEquals(offloader.offloadedLedgers().size(), 0);
+ } else if (sizeThreshold == null && timeThreshold == null) {
+ // Offloading is disabled, no ledgers will be offloaded.
+ assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 0);
+ assertEquals(offloader.offloadedLedgers().size(), 0);
+ } else if (sizeThreshold != null && sizeThreshold == 1L && timeThreshold != null && timeThreshold == 1L) {
+ // All the inactive ledgers will be offloaded
+ assertEventuallyTrue(() -> offloader.offloadedLedgers().size() == 2);
+ List<Long> allLedgerIds = ledger.getLedgersInfoAsList().stream().map(LedgerInfo::getLedgerId).toList();
+ assertEquals(offloader.offloadedLedgers(), Set.of(allLedgerIds.get(0), allLedgerIds.get(1)));
+ }
}
- @Test
- public void manualTriggerWhileAutoInProgress() throws Exception {
+ @DataProvider(name = "manualTriggerWhileAutoInProgress")
+ public Object[][] manualTriggerWhileAutoInProgressProvider() {
Review Comment:
Manually trigger offload is not controlled by size and time. It calculates the message id on the client side and then uses the message id to offload.
So this test is useless. And we need to support the time on the client side or support the related REST API.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org