You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "void-ptr974 (via GitHub)" <gi...@apache.org> on 2026/09/21 11:54:07 UTC
[PR] [fix][meta] Fix underreplication lock cleanup on close [pulsar]
void-ptr974 opened a new pull request, #26674:
URL: https://github.com/apache/pulsar/pull/26674
### Motivation
`PulsarLedgerUnderreplicationManager.close()` catches delete failures outside the loop over held locks. If one lock has already disappeared, it returns successfully without releasing the remaining locks. Other delete errors also stop cleanup at the first failed lock. When the metadata store is shared and remains open, those locks can continue blocking other replication workers.
Successfully deleted locks also remain in `heldLocks`. If another manager reacquires one of those locks, a second `close()` on the original manager can delete the new lock.
### Modifications
- Handle each lock deletion independently. Treat a missing lock as released and continue processing the remaining locks.
- Remove a lock from `heldLocks` only after a successful delete or `NotFoundException`, using conditional removal of the same lock entry.
- Continue cleanup after metadata-store errors or timeouts, preserve all failures in the thrown `UnavailableException`, and retain unsuccessful entries for a subsequent close attempt.
- Preserve the thread interrupt flag and stop promptly when a delete wait is interrupted, retaining any earlier failure as a suppressed exception.
- Add regressions for missing first/middle locks, repeated close after another manager reacquires released locks, multiple delete failures, timeout handling, and interruption after an earlier failure.
The change is limited to cleanup of tracked locks. It does not introduce ownership fencing for locks lost and recreated after session expiration or an ambiguous delete timeout. The existing per-delete timeout remains in effect; attempting all remaining locks can increase total close duration when several deletes time out.
### Verifying this change
Before the fix, all 12 new regression cases failed on Memory and real ZooKeeper for the expected cleanup/repeated-close assertions. Automatic test retries were disabled.
Post-fix validation, with automatic test retries disabled and no failed or skipped cases:
- macOS / JDK 25: 44 cases passed across Memory, real ZooKeeper, MockZooKeeper and RocksDB: 24 new regression cases plus 20 existing locking, release, completion, record-update and multiple-manager cases.
- Linux x86_64 / JDK 21 / Docker on home3: all 117 cases passed across all five providers, including real Oxia: 30 new regression cases and all 87 existing underreplication manager cases.
- `./gradlew quickCheck sanityCheck --max-workers=4 --console=plain` passed, covering repository-wide license/style checks and the main/test compilation checks provided by `sanityCheck`.
- The production and regression-test files used on home3 match the local files by SHA-256.
Metadata-store failures are injected at the delete API boundary; the timeout case injects a timed `Future.get()` failure without waiting 30 seconds. Missing-node tests delete real nodes, and repeated-close tests use real lock acquisition. The interruption test uses an interrupted thread waiting on an incomplete future.
- [ ] Make sure that the change passes the CI checks.
Full underreplication suites on Linux with Docker:
```bash
TEST_METADATA_PROVIDERS=Memory,ZooKeeper,MockZooKeeper,RocksDB,Oxia ./gradlew :pulsar-metadata:test \
--tests 'org.apache.pulsar.metadata.bookkeeper.PulsarLedgerUnderreplicationManagerCloseTest' \
--tests 'org.apache.pulsar.metadata.bookkeeper.LedgerUnderreplicationManagerTest' \
-PtestRetryCount=0 -PtestFailFast=false --max-workers=4 --console=plain
```
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#discussion_r4155222721
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -690,23 +690,35 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
public void close() throws ReplicationException.UnavailableException {
log.debug("close()");
notificationCallbackExecutor.shutdownNow();
- try {
- for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ ReplicationException.UnavailableException failure = null;
+ for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ try {
store.delete(e.getValue().getLockPath(), Optional.empty())
.get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
Review Comment:
Thanks for catching both points. I merged the latest `master` and changed `close()` to submit every tracked delete before waiting once on the aggregate with a single `BLOCKING_CALL_TIMEOUT`. This keeps the shutdown wait bounded while still attempting every lock. Successful and `NotFound` completions remove only the same tracked `Lock`; failures and pending results remain available for a later retry, and a late successful completion can still clear its original entry.
I added coverage for the single overall timeout, mixed success/failure/pending results, interruption, late successful completion, and repeated close through the explicit acquisition path from #26670.
The timeout/interruption result is still ambiguous at the metadata-store ownership boundary, so this change does not treat conditional removal as full fencing. I saw that you followed up with #26794 for ownership-aware cleanup in both `close()` and `releaseUnderreplicatedLedger()`. Thanks for taking that on; I linked it from this PR's description.
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#issuecomment-5931197558
Thanks @Denovo1998 for flagging the #26670 overlap. I merged the current `master`, including its `Lock`/`heldLocks` changes, and added a repeated-close regression using an explicitly acquired lock.
The updated targeted suite passed all 36 cases across Memory, ZooKeeper, MockZooKeeper, and RocksDB with retries disabled, and the metadata main/test checkstyle tasks passed. Oxia could not run locally because Docker is unavailable here, so that provider remains covered by CI. I also updated the PR description to reflect the shared shutdown timeout and the #26794 ownership-fencing follow-up.
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #26674:
URL: https://github.com/apache/pulsar/pull/26674
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#issuecomment-6292045091
This PR introduced a flaky test. Fixed in #27013. Please review
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#discussion_r4260241023
##########
pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * 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.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);
+ Optional<Long> version = invocation.getArgument(1);
+ if (attempts.getAndIncrement() == missingPosition) {
+ // Remove the actual node before the manager's delete, without relying on map iteration order.
+ return store.delete(path, version).thenCompose(__ -> store.delete(path, version));
+ }
+ return store.delete(path, version);
+ }).when(observed).delete(anyString(), any());
+
+ 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();
+ doAnswer(invocation -> {
+ attempts.incrementAndGet();
+ store.delete(invocation.getArgument(0), invocation.getArgument(1)).get(5, TimeUnit.SECONDS);
+ return delayedAcknowledgement;
+ }).when(observed).delete(anyString(), any());
+
+ assertThatThrownBy(() -> owner.close(100, TimeUnit.MILLISECONDS))
+ .isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
+ other.acquireUnderreplicatedLedger(ledgerId);
+ delayedAcknowledgement.complete(null);
Review Comment:
**[QUALITY] cover a late ack after the same manager re-acquires the ledger**
This test re-acquires through `other`, so it can't tell `heldLocks.remove(key, value)` from `remove(key)`. Please also cover the case where `owner` re-acquires the same ledger (a new `Lock` entry) before the late acknowledgement arrives. The new entry should survive.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -691,26 +694,77 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
@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()) {
+ deleteResults.add(FutureUtil.supplySafely(
+ () -> store.delete(entry.getValue().getLockPath(), Optional.empty()))
+ .handle((__, error) -> {
+ Throwable cause = error == null ? null : FutureUtil.unwrapCompletionException(error);
+ if (cause == null || cause instanceof MetadataStoreException.NotFoundException) {
+ heldLocks.remove(entry.getKey(), entry.getValue());
+ return null;
+ }
+ return new ExecutionException(cause);
Review Comment:
**[QUALITY] log failed lock deletions with the lock path**
The old `close()` logged `Error deleting underreplicated ledger lock`. Now a failure only becomes a value here. A failure that completes after the timeout or interrupt is never reported, and the exceptions don't say which lock failed.
Could you add a `log.warn()` with the lock path and ledger id in this branch? It would also cover late failures.
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#discussion_r4155222721
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -690,23 +690,35 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
public void close() throws ReplicationException.UnavailableException {
log.debug("close()");
notificationCallbackExecutor.shutdownNow();
- try {
- for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ ReplicationException.UnavailableException failure = null;
+ for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ try {
store.delete(e.getValue().getLockPath(), Optional.empty())
.get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
Review Comment:
Thanks for the review. I updated close() to submit all tracked deletes first and wait once with a single timeout. Successful and NotFound deletions remove only the matching tracked lock, while failures remain for retry. I also added tests for timeout, interruption, late completion, and repeated close. Ownership-aware cleanup is tracked in #26794.
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#issuecomment-6164407673
@lhotari Thanks for the review! Both inline comments are addressed in 5e150d1719f. Either merge order is fine; I’ll handle integration in whichever PR lands second.
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#discussion_r4123223767
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -690,23 +690,35 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
public void close() throws ReplicationException.UnavailableException {
log.debug("close()");
notificationCallbackExecutor.shutdownNow();
- try {
- for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ ReplicationException.UnavailableException failure = null;
+ for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ try {
store.delete(e.getValue().getLockPath(), Optional.empty())
.get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
Review Comment:
The ownership-fencing limitation looks reasonable to keep out of this PR, but I think the interrupted path has the same ambiguity as the timeout path.
Once `store.delete()` has been submitted, `get()` timing out or being interrupted doesn't mean the delete didn't complete. If the lock is recreated by another owner before a later `close()`, the retained stale entry can still delete that new lock because the retry uses `Optional.empty()`.
This also applies to `releaseUnderreplicatedLedger()`. Could we track this as a follow-up for both paths? I don't think it needs to block the cleanup fix here, but it would be good not to treat interruption as an unambiguous failed delete.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -690,23 +690,35 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
public void close() throws ReplicationException.UnavailableException {
log.debug("close()");
notificationCallbackExecutor.shutdownNow();
- try {
- for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ ReplicationException.UnavailableException failure = null;
+ for (Map.Entry<Long, Lock> e : heldLocks.entrySet()) {
+ try {
store.delete(e.getValue().getLockPath(), Optional.empty())
.get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);
Review Comment:
Could we avoid applying the full blocking timeout to every lock here?
`BLOCKING_CALL_TIMEOUT` is 30 seconds, and `ReplicationWorker` can have multiple locks in `heldLocks` while failed-ledger releases are deferred. During shutdown those timer tasks are cancelled and `underreplicationManager.close()` is used to clean up the remaining locks.
With this loop, if the metadata store stops responding, shutdown can now take `heldLocks.size() * 30s`. The old implementation would stop after the first timeout.
Would it make sense to use one overall deadline for `close()`, or issue the deletes first and wait for them with one bounded timeout? That should still let us attempt cleanup for every lock without making the shutdown time grow linearly with the number of held locks.
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#discussion_r4270209685
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -691,26 +694,77 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
@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()) {
+ deleteResults.add(FutureUtil.supplySafely(
+ () -> store.delete(entry.getValue().getLockPath(), Optional.empty()))
+ .handle((__, error) -> {
+ Throwable cause = error == null ? null : FutureUtil.unwrapCompletionException(error);
+ if (cause == null || cause instanceof MetadataStoreException.NotFoundException) {
+ heldLocks.remove(entry.getKey(), entry.getValue());
+ return null;
+ }
+ return new ExecutionException(cause);
Review Comment:
Thanks, this is what I had in mind. The warn is in the callback, so it also reports failures that arrive after the timeout.
##########
pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * 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.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);
+ Optional<Long> version = invocation.getArgument(1);
+ if (attempts.getAndIncrement() == missingPosition) {
+ // Remove the actual node before the manager's delete, without relying on map iteration order.
+ return store.delete(path, version).thenCompose(__ -> store.delete(path, version));
+ }
+ return store.delete(path, version);
+ }).when(observed).delete(anyString(), any());
+
+ 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();
+ doAnswer(invocation -> {
+ attempts.incrementAndGet();
+ store.delete(invocation.getArgument(0), invocation.getArgument(1)).get(5, TimeUnit.SECONDS);
+ return delayedAcknowledgement;
+ }).when(observed).delete(anyString(), any());
+
+ assertThatThrownBy(() -> owner.close(100, TimeUnit.MILLISECONDS))
+ .isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
+ other.acquireUnderreplicatedLedger(ledgerId);
+ delayedAcknowledgement.complete(null);
Review Comment:
Thanks, this covers the same-manager case. If the late acknowledgement dropped the new entry, release would skip the delete and the last assertion would fail. So the test tells the two removes apart.
--
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
Re: [PR] [fix][meta] Fix underreplication locks left behind when closing by handling each deletion independently [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26674:
URL: https://github.com/apache/pulsar/pull/26674#discussion_r4268578224
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -691,26 +694,77 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
@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()) {
+ deleteResults.add(FutureUtil.supplySafely(
+ () -> store.delete(entry.getValue().getLockPath(), Optional.empty()))
+ .handle((__, error) -> {
+ Throwable cause = error == null ? null : FutureUtil.unwrapCompletionException(error);
+ if (cause == null || cause instanceof MetadataStoreException.NotFoundException) {
+ heldLocks.remove(entry.getKey(), entry.getValue());
+ return null;
+ }
+ return new ExecutionException(cause);
Review Comment:
Added WARN logging with the ledger ID, lock path, and exception. This also covers failures arriving after timeout or interruption.
##########
pulsar-metadata/src/test/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManagerCloseTest.java:
##########
@@ -0,0 +1,415 @@
+/*
+ * 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.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);
+ Optional<Long> version = invocation.getArgument(1);
+ if (attempts.getAndIncrement() == missingPosition) {
+ // Remove the actual node before the manager's delete, without relying on map iteration order.
+ return store.delete(path, version).thenCompose(__ -> store.delete(path, version));
+ }
+ return store.delete(path, version);
+ }).when(observed).delete(anyString(), any());
+
+ 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();
+ doAnswer(invocation -> {
+ attempts.incrementAndGet();
+ store.delete(invocation.getArgument(0), invocation.getArgument(1)).get(5, TimeUnit.SECONDS);
+ return delayedAcknowledgement;
+ }).when(observed).delete(anyString(), any());
+
+ assertThatThrownBy(() -> owner.close(100, TimeUnit.MILLISECONDS))
+ .isInstanceOf(UnavailableException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
+ other.acquireUnderreplicatedLedger(ledgerId);
+ delayedAcknowledgement.complete(null);
Review Comment:
Added a same-manager reacquisition test verifying that the replacement lock remains tracked after the old acknowledgement arrives. It fails with `remove(key)` and passes with `remove(key, value)`. Local `quickCheck` and all 40 cases across four providers passed.
--
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