You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2023/12/07 20:21:03 UTC
(pulsar) branch master updated: [fix][offload] Don't cleanup data when offload met MetaStore exception (#21686)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 1ccf90d95a7 [fix][offload] Don't cleanup data when offload met MetaStore exception (#21686)
1ccf90d95a7 is described below
commit 1ccf90d95a7e7d291b8d27f627d06ac7da607954
Author: Yong Zhang <zh...@gmail.com>
AuthorDate: Fri Dec 8 04:20:55 2023 +0800
[fix][offload] Don't cleanup data when offload met MetaStore exception (#21686)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 3 +-
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 55 ++++++++++++++++++++++
2 files changed, 57 insertions(+), 1 deletion(-)
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 1e5d67871d4..8ce2a6924eb 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
@@ -3209,7 +3209,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
}
- private void offloadLoop(CompletableFuture<PositionImpl> promise, Queue<LedgerInfo> ledgersToOffload,
+ void offloadLoop(CompletableFuture<PositionImpl> promise, Queue<LedgerInfo> ledgersToOffload,
PositionImpl firstUnoffloaded, Optional<Throwable> firstError) {
State currentState = getState();
if (currentState == State.Closed) {
@@ -3257,6 +3257,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
log.error("[{}] Failed to update offloaded metadata for the ledgerId {}, "
+ "the offloaded data will not be cleaned up",
name, ledgerId, exception);
+ return;
} else {
log.error("[{}] Failed to offload data for the ledgerId {}, "
+ "clean up the offloaded data",
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 601cc4633d7..8430afb4e4f 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
@@ -21,7 +21,9 @@ package org.apache.bookkeeper.mledger.impl;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
@@ -55,10 +57,12 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Optional;
+import java.util.Queue;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
@@ -4152,4 +4156,55 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
}
return taskCounter;
}
+
+ @Test
+ public void testNoCleanupOffloadLedgerWhenMetadataExceptionHappens() throws Exception {
+ ManagedLedgerConfig config = spy(new ManagedLedgerConfig());
+ ManagedLedgerImpl ml = spy((ManagedLedgerImpl) factory.open("testNoCleanupOffloadLedger", config));
+
+ // mock the ledger offloader
+ LedgerOffloader ledgerOffloader = mock(NullLedgerOffloader.class);
+ when(config.getLedgerOffloader()).thenReturn(ledgerOffloader);
+ when(ledgerOffloader.getOffloadDriverName()).thenReturn("mock");
+
+ // There will have two put call to the metadata store, the first time is prepare the offload.
+ // And the second is the complete the offload. This case is testing when completing the offload,
+ // the metadata store meets an exception.
+ AtomicInteger metadataPutCallCount = new AtomicInteger(0);
+ metadataStore.failConditional(new MetadataStoreException("mock completion error"),
+ (key, value) -> key.equals(FaultInjectionMetadataStore.OperationType.PUT) &&
+ metadataPutCallCount.incrementAndGet() == 2);
+
+ // prepare the arguments for the offloadLoop method
+ CompletableFuture<PositionImpl> future = new CompletableFuture<>();
+ Queue<LedgerInfo> ledgersToOffload = new LinkedList<>();
+ LedgerInfo ledgerInfo = LedgerInfo.getDefaultInstance().toBuilder().setLedgerId(1).setEntries(10).build();
+ ledgersToOffload.add(ledgerInfo);
+ PositionImpl firstUnoffloaded = new PositionImpl(1, 0);
+ Optional<Throwable> firstError = Optional.empty();
+
+ // mock the read handle to make the offload successful
+ CompletableFuture<ReadHandle> readHandle = new CompletableFuture<>();
+ readHandle.complete(mock(ReadHandle.class));
+ when(ml.getLedgerHandle(eq(ledgerInfo.getLedgerId()))).thenReturn(readHandle);
+ when(ledgerOffloader.offload(any(), any(), anyMap())).thenReturn(CompletableFuture.completedFuture(null));
+
+ ml.ledgers.put(ledgerInfo.getLedgerId(), ledgerInfo);
+
+ // do the offload
+ ml.offloadLoop(future, ledgersToOffload, firstUnoffloaded, firstError);
+
+ // waiting for the offload complete
+ try {
+ future.join();
+ fail("The offload should fail");
+ } catch (Exception e) {
+ // the offload should fail
+ assertTrue(e.getCause().getMessage().contains("mock completion error"));
+ }
+
+ // the ledger deletion shouldn't happen
+ verify(ledgerOffloader, times(0))
+ .deleteOffloaded(eq(ledgerInfo.getLedgerId()), any(), anyMap());
+ }
}