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 2026/10/17 20:20:09 UTC
(pulsar) 15/22: [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently (#26674)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e16041361b534aef3d84c67bc9210fdf5faa61fc
Author: void-ptr974 <vo...@protonmail.com>
AuthorDate: Sat Oct 17 23:17:49 2026 +0800
[fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently (#26674)
Co-authored-by: Matteo Merli <mm...@apache.org>
(cherry picked from commit ecdd47d51257b384741734dfe7f332afbfbc7698)
---
.../PulsarLedgerUnderreplicationManager.java | 81 +++-
...lsarLedgerUnderreplicationManagerCloseTest.java | 467 +++++++++++++++++++++
2 files changed, 535 insertions(+), 13 deletions(-)
diff --git a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java
index 410e2c98e661..371b3521ea2b 100644
--- a/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java
+++ b/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java
@@ -21,6 +21,7 @@ package org.apache.pulsar.metadata.bookkeeper;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.pulsar.metadata.bookkeeper.AbstractMetadataDriver.BLOCKING_CALL_TIMEOUT;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.UnknownHostException;
@@ -40,6 +41,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
@@ -772,25 +774,78 @@ public class PulsarLedgerUnderreplicationManager implements LedgerUnderreplicati
@Override
public void close() throws ReplicationException.UnavailableException {
+ close(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
+ }
+
+ @VisibleForTesting
+ void close(long timeout, TimeUnit timeoutUnit) throws ReplicationException.UnavailableException {
log.debug("close()");
notificationCallbackExecutor.shutdownNow();
+
+ Map<Long, Lock> locks = Map.copyOf(heldLocks);
+ List<CompletableFuture<Throwable>> deleteResults = new ArrayList<>(locks.size());
+ for (Map.Entry<Long, Lock> entry : locks.entrySet()) {
+ // releaseLock() verifies ownership before deleting and drops the held lock once it is gone.
+ deleteResults.add(FutureUtil.supplySafely(() -> releaseLock(entry.getKey(), entry.getValue()))
+ .handle((__, error) -> {
+ Throwable cause = error == null ? null : FutureUtil.unwrapCompletionException(error);
+ if (cause == null || cause instanceof MetadataStoreException.NotFoundException) {
+ return null;
+ }
+ log.warn().attr("ledgerId", entry.getKey()).attr("lockPath", entry.getValue().getLockPath())
+ .exception(cause).log("Error deleting underreplicated ledger lock");
+ return new ExecutionException(cause);
+ }));
+ }
+
try {
- for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
- releaseLock(e.getKey(), e.getValue()).get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
- }
- } catch (ExecutionException ee) {
- if (ee.getCause() instanceof MetadataStoreException.NotFoundException) {
- // this is ok
- } else {
- log.error().exception(ee).log("Error deleting underreplicated ledger lock");
- throw new ReplicationException.UnavailableException("Error contacting metadata store", ee);
+ FutureUtil.waitForAll(deleteResults).get(timeout, timeoutUnit);
+ } catch (TimeoutException timeoutException) {
+ var timeoutFailure = new ReplicationException.UnavailableException(
+ "Timed out while deleting underreplicated ledger locks", timeoutException);
+ ReplicationException.UnavailableException completedFailures = collectDeleteFailures(deleteResults);
+ if (completedFailures != null) {
+ timeoutFailure.addSuppressed(completedFailures);
}
- } catch (TimeoutException ex) {
- throw new ReplicationException.UnavailableException("Error contacting metadata store", ex);
- } catch (InterruptedException ie) {
+ throw timeoutFailure;
+ } catch (InterruptedException interruptedException) {
Thread.currentThread().interrupt();
- throw new ReplicationException.UnavailableException("Interrupted while connecting metadata store", ie);
+ var interruptedFailure = new ReplicationException.UnavailableException(
+ "Interrupted while deleting underreplicated ledger locks", interruptedException);
+ ReplicationException.UnavailableException completedFailures = collectDeleteFailures(deleteResults);
+ if (completedFailures != null) {
+ interruptedFailure.addSuppressed(completedFailures);
+ }
+ throw interruptedFailure;
+ } catch (ExecutionException executionException) {
+ throw new ReplicationException.UnavailableException(
+ "Error deleting underreplicated ledger locks", executionException);
+ }
+
+ ReplicationException.UnavailableException failure = collectDeleteFailures(deleteResults);
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ private static ReplicationException.UnavailableException collectDeleteFailures(
+ Collection<CompletableFuture<Throwable>> deleteResults) {
+ ReplicationException.UnavailableException failure = null;
+ for (CompletableFuture<Throwable> deleteResult : deleteResults) {
+ if (!deleteResult.isDone()) {
+ continue;
+ }
+ Throwable error = deleteResult.getNow(null);
+ if (error != null) {
+ if (failure == null) {
+ failure = new ReplicationException.UnavailableException(
+ "Error contacting metadata store", error);
+ } else {
+ failure.addSuppressed(error);
+ }
+ }
}
+ return failure;
}
@Override
diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
new file mode 100644
index 000000000000..e41f12942eac
--- /dev/null
+++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.metadata.bookkeeper;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.spy;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
+import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.metadata.BaseMetadataStoreTest;
+import org.apache.pulsar.metadata.api.MetadataStoreConfig;
+import org.apache.pulsar.metadata.api.MetadataStoreException;
+import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class PulsarLedgerUnderreplicationManagerCloseTest extends BaseMetadataStoreTest {
+ private static final List<Long> LEDGERS = List.of(1L, 2L, 3L);
+
+ @DataProvider(name = "implWithMissingPosition")
+ public Object[][] implementationsWithMissingPosition() {
+ List<Object[]> cases = new ArrayList<>();
+ for (Object[] implementation : implementations()) {
+ cases.add(new Object[]{implementation[0], implementation[1], 0});
+ cases.add(new Object[]{implementation[0], implementation[1], 1});
+ }
+ return cases.toArray(Object[][]::new);
+ }
+
+ @Test(dataProvider = "implWithMissingPosition", timeOut = 30000)
+ public void testCloseContinuesAfterMissingLock(String provider, Supplier<String> urlSupplier, int missingPosition)
+ throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ try (var store = newStore(urlSupplier)) {
+ var observed = spy(store);
+ try (var owner = newManager(observed, root);
+ var other = newManager(store, root)) {
+ acquireLedgers(owner);
+ AtomicInteger attempts = new AtomicInteger();
+ doAnswer(invocation -> {
+ String path = invocation.getArgument(0);
+ if (attempts.getAndIncrement() == missingPosition) {
+ // Remove the actual node before the manager's ownership read, without relying on map
+ // iteration order.
+ return store.delete(path, Optional.empty()).thenCompose(__ -> store.get(path));
+ }
+ return store.get(path);
+ }).when(observed).get(anyString());
+
+ owner.close();
+
+ assertThat(attempts.get()).isEqualTo(LEDGERS.size());
+ assertAllUnlocked(other);
+ acquireExistingLedgers(other);
+ owner.close();
+ assertThat(attempts.get()).isEqualTo(LEDGERS.size());
+ assertAllLocked(other);
+ }
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testRepeatedCloseDoesNotDeleteReacquiredLocks(String provider, Supplier<String> urlSupplier)
+ throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ try (var store = newStore(urlSupplier);
+ var owner = newManager(store, root);
+ var other = newManager(store, root)) {
+ acquireLedgers(owner);
+ owner.close();
+ assertAllUnlocked(other);
+ acquireExistingLedgers(other);
+
+ owner.close();
+
+ assertAllLocked(other);
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testRepeatedCloseDoesNotDeleteExplicitlyReacquiredLock(
+ String provider, Supplier<String> urlSupplier) throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ long ledgerId = 1L;
+ try (var store = newStore(urlSupplier);
+ var owner = newManager(store, root);
+ var other = newManager(store, root)) {
+ owner.acquireUnderreplicatedLedger(ledgerId);
+ owner.close();
+ assertThat(other.isLedgerBeingReplicated(ledgerId)).isFalse();
+ other.acquireUnderreplicatedLedger(ledgerId);
+
+ owner.close();
+
+ assertThat(other.isLedgerBeingReplicated(ledgerId)).isTrue();
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testCloseAggregatesFailuresAndRetriesOnlyUnreleasedLocks(String provider, Supplier<String> urlSupplier)
+ throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ try (var store = newStore(urlSupplier)) {
+ var observed = spy(store);
+ try (var owner = newManager(observed, root);
+ var other = newManager(store, root)) {
+ acquireLedgers(owner);
+ List<String> attemptedPaths = new ArrayList<>();
+ List<MetadataStoreException> failures = List.of(new MetadataStoreException("first failure"),
+ new MetadataStoreException("second failure"));
+ doAnswer(invocation -> {
+ String path = invocation.getArgument(0);
+ attemptedPaths.add(path);
+ if (attemptedPaths.size() <= failures.size()) {
+ return CompletableFuture.failedFuture(failures.get(attemptedPaths.size() - 1));
+ }
+ return store.delete(path, invocation.getArgument(1));
+ }).when(observed).delete(anyString(), any());
+
+ assertThatThrownBy(owner::close).isInstanceOf(UnavailableException.class)
+ .satisfies(error -> {
+ assertThat(error.getCause()).isInstanceOf(ExecutionException.class)
+ .hasCause(failures.get(0));
+ assertThat(error.getSuppressed()).hasSize(1);
+ assertThat(error.getSuppressed()[0]).isInstanceOf(ExecutionException.class)
+ .hasCause(failures.get(1));
+ });
+ assertThat(attemptedPaths).hasSize(LEDGERS.size());
+ for (long ledgerId : LEDGERS) {
+ assertThat(other.isLedgerBeingReplicated(ledgerId))
+ .isEqualTo(attemptedPaths.subList(0, 2).contains(lockPath(root, ledgerId)));
+ }
+ long reacquired = other.pollLedgerToRereplicate();
+ assertThat(reacquired).isIn(LEDGERS);
+ List<String> retryPaths = new ArrayList<>();
+ doAnswer(invocation -> {
+ String path = invocation.getArgument(0);
+ retryPaths.add(path);
+ return store.delete(path, invocation.getArgument(1));
+ }).when(observed).delete(anyString(), any());
+
+ owner.close();
+
+ assertThat(retryPaths).containsExactlyInAnyOrderElementsOf(attemptedPaths.subList(0, 2));
+ assertThat(other.isLedgerBeingReplicated(reacquired)).isTrue();
+ }
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testCloseContinuesAfterTimeout(String provider, Supplier<String> urlSupplier) throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ try (var store = newStore(urlSupplier)) {
+ var observed = spy(store);
+ try (var owner = newManager(observed, root);
+ var other = newManager(store, root)) {
+ acquireLedgers(owner);
+ CompletableFuture<Void> pendingDelete = new CompletableFuture<>();
+ MetadataStoreException immediateFailure = new MetadataStoreException("immediate failure");
+ List<CompletableFuture<Void>> completedDeletes = new ArrayList<>();
+ List<String> attemptedPaths = new ArrayList<>();
+ doAnswer(invocation -> {
+ String path = invocation.getArgument(0);
+ attemptedPaths.add(path);
+ if (attemptedPaths.size() == 1) {
+ return pendingDelete;
+ }
+ if (attemptedPaths.size() == 2) {
+ return CompletableFuture.failedFuture(immediateFailure);
+ }
+ CompletableFuture<Void> delete = store.delete(path, invocation.getArgument(1));
+ completedDeletes.add(delete);
+ return delete;
+ }).when(observed).delete(anyString(), any());
+
+ assertThatThrownBy(() -> owner.close(100, TimeUnit.MILLISECONDS))
+ .isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(TimeoutException.class)
+ .satisfies(error -> {
+ assertThat(error.getSuppressed()).hasSize(1);
+ assertThat(error.getSuppressed()[0]).hasRootCause(immediateFailure);
+ });
+ FutureUtil.waitForAll(completedDeletes).get(5, TimeUnit.SECONDS);
+ assertThat(attemptedPaths).hasSize(LEDGERS.size());
+ for (long ledgerId : LEDGERS) {
+ assertThat(other.isLedgerBeingReplicated(ledgerId))
+ .isEqualTo(attemptedPaths.subList(0, 2).contains(lockPath(root, ledgerId)));
+ }
+
+ pendingDelete.completeExceptionally(new MetadataStoreException("delete failed"));
+ List<String> retryPaths = new ArrayList<>();
+ doAnswer(invocation -> {
+ retryPaths.add(invocation.getArgument(0));
+ return store.delete(invocation.getArgument(0), invocation.getArgument(1));
+ })
+ .when(observed).delete(anyString(), any());
+ owner.close();
+
+ assertThat(retryPaths).containsExactlyInAnyOrderElementsOf(attemptedPaths.subList(0, 2));
+ assertAllUnlocked(other);
+ }
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testCloseUsesSingleOverallTimeout(String provider, Supplier<String> urlSupplier) throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ int lockCount = 20;
+ try (var store = newStore(urlSupplier)) {
+ var observed = spy(store);
+ try (var owner = newManager(observed, root)) {
+ for (long ledgerId = 1; ledgerId <= lockCount; ledgerId++) {
+ owner.acquireUnderreplicatedLedger(ledgerId);
+ }
+ AtomicInteger attempts = new AtomicInteger();
+ List<CompletableFuture<Void>> pendingDeletes = new ArrayList<>();
+ doAnswer(invocation -> {
+ attempts.incrementAndGet();
+ CompletableFuture<Void> pendingDelete = new CompletableFuture<>();
+ pendingDeletes.add(pendingDelete);
+ return pendingDelete;
+ }).when(observed).delete(anyString(), any());
+
+ long startNanos = System.nanoTime();
+ try {
+ assertThatThrownBy(() -> owner.close(100, TimeUnit.MILLISECONDS))
+ .isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
+ long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ assertThat(attempts.get()).isEqualTo(lockCount);
+ assertThat(elapsedMillis).isLessThan(1000L);
+ } finally {
+ pendingDeletes.forEach(delete -> delete.completeExceptionally(
+ new MetadataStoreException("delete failed")));
+ doAnswer(invocation -> store.delete(invocation.getArgument(0), invocation.getArgument(1)))
+ .when(observed).delete(anyString(), any());
+ owner.close();
+ }
+ }
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testLateSuccessfulDeleteIsNotRetriedAfterTimeout(
+ String provider, Supplier<String> urlSupplier) throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ long ledgerId = 1L;
+ try (var store = newStore(urlSupplier)) {
+ var observed = spy(store);
+ try (var owner = newManager(observed, root);
+ var other = newManager(store, root)) {
+ owner.acquireUnderreplicatedLedger(ledgerId);
+ CompletableFuture<Void> delayedAcknowledgement = new CompletableFuture<>();
+ AtomicInteger attempts = new AtomicInteger();
+ CompletableFuture<Void> actualDelete = new CompletableFuture<>();
+ doAnswer(invocation -> {
+ attempts.incrementAndGet();
+ // The delete runs on the store's callback thread, so chain on it instead of blocking.
+ FutureUtil.completeAfter(actualDelete,
+ store.delete(invocation.getArgument(0), invocation.getArgument(1)));
+ return actualDelete.thenCompose(__ -> delayedAcknowledgement);
+ }).when(observed).delete(anyString(), any());
+
+ assertThatThrownBy(() -> owner.close(100, TimeUnit.MILLISECONDS))
+ .isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
+ actualDelete.get(5, TimeUnit.SECONDS);
+ other.acquireUnderreplicatedLedger(ledgerId);
+ delayedAcknowledgement.complete(null);
+
+ owner.close();
+
+ assertThat(attempts.get()).isEqualTo(1);
+ assertThat(other.isLedgerBeingReplicated(ledgerId)).isTrue();
+ }
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testLateSuccessfulDeletePreservesLockReacquiredBySameManager(
+ String provider, Supplier<String> urlSupplier) throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ long ledgerId = 1L;
+ String path = lockPath(root, ledgerId);
+ try (var store = newStore(urlSupplier)) {
+ var observed = spy(store);
+ try (var owner = newManager(observed, root)) {
+ owner.acquireUnderreplicatedLedger(ledgerId);
+ CompletableFuture<Void> delayedAcknowledgement = new CompletableFuture<>();
+ CompletableFuture<Void> actualDelete = new CompletableFuture<>();
+ doAnswer(invocation -> {
+ // The delete runs on the store's callback thread, so chain on it instead of blocking.
+ FutureUtil.completeAfter(actualDelete,
+ store.delete(invocation.getArgument(0), invocation.getArgument(1)));
+ return actualDelete.thenCompose(__ -> delayedAcknowledgement);
+ }).when(observed).delete(anyString(), any());
+
+ try {
+ assertThatThrownBy(() -> owner.close(100, TimeUnit.MILLISECONDS))
+ .isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
+ actualDelete.get(5, TimeUnit.SECONDS);
+ assertThat(store.get(path).get(5, TimeUnit.SECONDS))
+ .as("the original lock was deleted before its acknowledgement")
+ .isEmpty();
+ owner.acquireUnderreplicatedLedger(ledgerId);
+ assertThat(store.get(path).get(5, TimeUnit.SECONDS))
+ .as("the same manager reacquired the lock")
+ .isPresent();
+ delayedAcknowledgement.complete(null);
+ reset(observed);
+
+ owner.releaseUnderreplicatedLedger(ledgerId);
+
+ assertThat(store.get(path).get(5, TimeUnit.SECONDS))
+ .as("late acknowledgement preserves ownership of the replacement lock")
+ .isEmpty();
+ } finally {
+ reset(observed);
+ delayedAcknowledgement.completeExceptionally(new MetadataStoreException("delete failed"));
+ }
+ }
+ }
+ }
+
+ @Test(dataProvider = "impl", timeOut = 30000)
+ public void testClosePreservesInterruptAndPreviousFailure(String provider, Supplier<String> urlSupplier)
+ throws Exception {
+ String root = "/ledgers-" + UUID.randomUUID();
+ try (var store = newStore(urlSupplier)) {
+ var observed = spy(store);
+ try (var owner = newManager(observed, root);
+ var other = newManager(store, root)) {
+ acquireLedgers(owner);
+ MetadataStoreException firstFailure = new MetadataStoreException("first failure");
+ AtomicInteger attempts = new AtomicInteger();
+ CountDownLatch allDeletesSubmitted = new CountDownLatch(LEDGERS.size());
+ List<CompletableFuture<Void>> pendingDeletes = new ArrayList<>();
+ doAnswer(invocation -> {
+ int attempt = attempts.getAndIncrement();
+ CompletableFuture<Void> result;
+ if (attempt == 0) {
+ result = CompletableFuture.failedFuture(firstFailure);
+ } else {
+ result = new CompletableFuture<>();
+ pendingDeletes.add(result);
+ }
+ allDeletesSubmitted.countDown();
+ return result;
+ }).when(observed).delete(anyString(), any());
+ AtomicReference<Throwable> closeFailure = new AtomicReference<>();
+ AtomicBoolean interruptPreserved = new AtomicBoolean();
+ Thread closeThread = new Thread(() -> {
+ try {
+ owner.close();
+ } catch (UnavailableException error) {
+ closeFailure.set(error);
+ interruptPreserved.set(Thread.currentThread().isInterrupted());
+ }
+ });
+ try {
+ closeThread.start();
+ assertThat(allDeletesSubmitted.await(5, TimeUnit.SECONDS)).isTrue();
+ closeThread.interrupt();
+ closeThread.join(5000);
+ assertThat(closeThread.isAlive()).isFalse();
+ assertThat(closeFailure.get()).isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(InterruptedException.class)
+ .satisfies(error -> {
+ assertThat(error.getSuppressed()).hasSize(1);
+ assertThat(error.getSuppressed()[0]).isInstanceOf(UnavailableException.class)
+ .hasRootCause(firstFailure);
+ });
+ assertThat(interruptPreserved.get()).isTrue();
+ assertThat(attempts.get()).isEqualTo(LEDGERS.size());
+ } finally {
+ if (closeThread.isAlive()) {
+ closeThread.interrupt();
+ closeThread.join(5000);
+ }
+ pendingDeletes.forEach(delete -> delete.completeExceptionally(
+ new MetadataStoreException("delete failed")));
+ doAnswer(invocation -> store.delete(invocation.getArgument(0), invocation.getArgument(1)))
+ .when(observed).delete(anyString(), any());
+ }
+ assertAllLocked(other);
+ owner.close();
+ assertAllUnlocked(other);
+ }
+ }
+ }
+
+ private static MetadataStoreExtended newStore(Supplier<String> urlSupplier) throws Exception {
+ return MetadataStoreExtended.create(urlSupplier.get(), MetadataStoreConfig.builder().build());
+ }
+
+ private static PulsarLedgerUnderreplicationManager newManager(MetadataStoreExtended store, String root)
+ throws Exception {
+ return new PulsarLedgerUnderreplicationManager(new ClientConfiguration(), store, root);
+ }
+
+ private static String lockPath(String root, long ledgerId) {
+ return PulsarLedgerUnderreplicationManager.getUrLedgerLockPath(
+ PulsarLedgerUnderreplicationManager.getUrLockPath(root), ledgerId);
+ }
+
+ private static void acquireLedgers(PulsarLedgerUnderreplicationManager manager) throws Exception {
+ for (long ledgerId : LEDGERS) {
+ manager.markLedgerUnderreplicated(ledgerId, "bookie:3181");
+ }
+ acquireExistingLedgers(manager);
+ }
+
+ private static void acquireExistingLedgers(PulsarLedgerUnderreplicationManager manager) throws Exception {
+ List<Long> acquired = new ArrayList<>();
+ for (int i = 0; i < LEDGERS.size(); i++) {
+ acquired.add(manager.pollLedgerToRereplicate());
+ }
+ assertThat(acquired).containsExactlyInAnyOrderElementsOf(LEDGERS);
+ }
+
+ private static void assertAllLocked(PulsarLedgerUnderreplicationManager manager) throws Exception {
+ for (long ledgerId : LEDGERS) {
+ assertThat(manager.isLedgerBeingReplicated(ledgerId)).as("ledger %s remains locked", ledgerId).isTrue();
+ }
+ }
+
+ private static void assertAllUnlocked(PulsarLedgerUnderreplicationManager manager) throws Exception {
+ for (long ledgerId : LEDGERS) {
+ assertThat(manager.isLedgerBeingReplicated(ledgerId)).as("ledger %s is unlocked", ledgerId).isFalse();
+ }
+ }
+}