You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2022/03/19 06:47:53 UTC
[pulsar] branch branch-2.9 updated: cancel offload tasks when managed ledger closed. (#14748)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new bf98fa7 cancel offload tasks when managed ledger closed. (#14748)
bf98fa7 is described below
commit bf98fa7c276542f64d3216ac4baa7c5069c1dd2f
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Sat Mar 19 14:46:09 2022 +0800
cancel offload tasks when managed ledger closed. (#14748)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 9 +++--
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 40 ++++++++++++++++++++++
2 files changed, 47 insertions(+), 2 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 59b04fe..193ce42 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -2298,13 +2298,13 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
+ ", 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);
}
-
- offloadLoop(unlockingPromise, toOffload, PositionImpl.latest, Optional.empty());
}
}
}
@@ -2862,6 +2862,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
private void offloadLoop(CompletableFuture<PositionImpl> promise, Queue<LedgerInfo> ledgersToOffload,
PositionImpl firstUnoffloaded, Optional<Throwable> firstError) {
+ if (getState() == State.Closed) {
+ promise.completeExceptionally(new ManagedLedgerAlreadyClosedException(
+ String.format("managed ledger [%s] has already closed", name)));
+ return;
+ }
LedgerInfo info = ledgersToOffload.poll();
if (info == null) {
if (firstError.isPresent()) {
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index 648d3e6..af5f6c8 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -54,6 +54,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
@@ -121,6 +122,7 @@ import org.apache.pulsar.common.policies.data.EnsemblePlacementPolicyConfig;
import org.apache.pulsar.metadata.api.MetadataStoreException;
import org.apache.pulsar.metadata.api.Stat;
import org.awaitility.Awaitility;
+import org.mockito.Mockito;
import org.testng.Assert;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -3422,4 +3424,42 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
assertTrue(timeoutTask2.isCancelled());
assertTrue(checkLedgerRollTask2.isCancelled());
}
+
+ @Test
+ public void testOffloadTaskCancelled() throws Exception {
+ ManagedLedgerFactory factory = new ManagedLedgerFactoryImpl(metadataStore, bkc);
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMaxEntriesPerLedger(2);
+ config.setMinimumRolloverTime(0, TimeUnit.SECONDS);
+
+ OffloadPoliciesImpl offloadPolicies = new OffloadPoliciesImpl();
+ offloadPolicies.setManagedLedgerOffloadDriver("mock");
+ offloadPolicies.setManagedLedgerOffloadThresholdInBytes(0L);
+ LedgerOffloader ledgerOffloader = Mockito.mock(LedgerOffloader.class);
+ Mockito.when(ledgerOffloader.getOffloadPolicies()).thenReturn(offloadPolicies);
+ Mockito.when(ledgerOffloader.getOffloadDriverName()).thenReturn(offloadPolicies.getManagedLedgerOffloadDriver());
+ config.setLedgerOffloader(ledgerOffloader);
+
+ CompletableFuture<ReadHandle> readHandle = new CompletableFuture<>();
+ readHandle.complete(mock(ReadHandle.class));
+
+ CompletableFuture<Void> offloadFuture = new CompletableFuture<>();
+ offloadFuture.complete(null);
+ Mockito.when(ledgerOffloader.offload(any(ReadHandle.class), any(UUID.class), any(Map.class))).thenReturn(offloadFuture);
+
+ final ManagedLedgerImpl ledgerInit = (ManagedLedgerImpl) factory.open("test-offload-task-close", config);
+ final ManagedLedgerImpl ledger = spy(ledgerInit);
+ long ledgerId = 3L;
+ doReturn(readHandle).when(ledger).getLedgerHandle(ledgerId);
+ doReturn(ManagedLedgerImpl.State.Closed).when(ledger).getState();
+ ledger.addEntry("dummy-entry-1".getBytes(Encoding));
+ ledger.addEntry("dummy-entry-2".getBytes(Encoding));
+ ledger.addEntry("dummy-entry-3".getBytes(Encoding));
+ ledger.close();
+
+ Awaitility.await().untilAsserted(() -> {
+ CompletableFuture<LedgerInfo> ledgerInfo = ledger.getLedgerInfo(ledgerId);
+ Assert.assertFalse(ledgerInfo.get(100, TimeUnit.MILLISECONDS).getOffloadContext().getComplete());
+ });
+ }
}