You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Denovo1998 (via GitHub)" <gi...@apache.org> on 2026/10/01 03:58:19 UTC
[PR] [fix][meta] Preserve ownership during underreplication lock cleanup [pulsar]
Denovo1998 opened a new pull request, #26794:
URL: https://github.com/apache/pulsar/pull/26794
### Motivation
Underreplication locks can survive session-loss notifications and metadata-store migration. Losing their local ownership records or treating failed cleanup as successful can leave orphaned locks that block ledger rereplication. Cleanup must retain ownership until the acquisition is safely released and preserve locks belonging to another acquisition.
### Modifications
- Identify each lock acquisition with a UUID stored in a protobuf text-format comment, preserving compatibility with existing BookKeeper readers.
- Verify the tracked acquisition and metadata-store ownership before deleting a lock or its underreplication record.
- Use the current lock version for conditional deletion.
- Share in-flight cleanup futures and retain ownership records after failed cleanup so subsequent attempts can retry.
- Preserve successful acquisitions across session-loss notifications and return retryable cleanup failures when metadata migration makes ownership uncertain.
- Extend `LedgerUnderreplicationManagerTest` with real-backend regressions for session recovery and expiration, ownership replacement, concurrent cleanup, version conflicts, migration retries, and source-session expiration.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
This change added tests and can be verified as follows:
```bash
TEST_METADATA_PROVIDERS=Memory,ZooKeeper,MockZooKeeper,RocksDB,Oxia \
./gradlew :pulsar-metadata:test \
--tests 'org.apache.pulsar.metadata.bookkeeper.LedgerUnderreplicationManagerTest' \
-PtestRetryCount=0 -PtestFailFast=false --max-workers=4
./gradlew quickCheck
```
Local validation passed all 165 tests with no failures, errors, or skipped tests. `quickCheck` also passed. The new regressions reproduced the cleanup failures before the corresponding fixes.
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies (add or upgrade a dependency)
- [x] The public API — adds `DualMetadataStore#getMigrationPhase()`
- [ ] 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] Preserve ownership during underreplication lock cleanup [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26794:
URL: https://github.com/apache/pulsar/pull/26794#discussion_r4198791291
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -686,7 +698,70 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted while connecting metadata store", ie);
}
- heldLocks.remove(ledgerId);
+ if (l != null) {
+ heldLocks.remove(ledgerId, l);
+ }
+ }
+
+ private CompletableFuture<Void> releaseLock(long ledgerId, Lock lock) {
+ if (heldLocks.get(ledgerId) != lock) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ CompletableFuture<Void> existingRelease = lock.releaseFuture.compareAndExchange(null, result);
+ if (existingRelease != null) {
+ return existingRelease;
+ }
+
+ // A timed-out or interrupted wait does not cancel the delete. Reuse its outcome instead
+ // of sending another delete, which could target a lock acquired by a different manager.
+ MigrationPhase migrationPhase = getStoreMigrationPhase();
+ FutureUtil.supplySafely(() -> {
+ if (migrationPhase == MigrationPhase.PREPARATION || migrationPhase == MigrationPhase.COPYING) {
+ return CompletableFuture.<Optional<GetResult>>failedFuture(
+ new MetadataStoreException("Metadata migration is in progress; retry lock cleanup"));
+ }
+ return store.get(lock.getLockPath());
+ })
+ .thenCompose(existing -> {
+ if (getStoreMigrationPhase() != migrationPhase) {
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Metadata migration changed during lock cleanup; retry"));
+ }
+ if (existing.isEmpty() || !ownsLock(ledgerId, lock, existing.get())) {
+ if (migrationPhase == MigrationPhase.FAILED) {
+ // A failed migration may still retain this acquisition in the inactive target.
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Lock ownership is uncertain after failed migration"));
+ }
+ return CompletableFuture.<Void>completedFuture(null);
+ }
+ return store.delete(lock.getLockPath(), Optional.of(existing.get().getStat().getVersion()));
Review Comment:
**[SUGGESTION] the version-conditional delete does not guard against a replacement lock on ZooKeeper, Memory and RocksDB**
The lock node is never updated after creation, so its version is always 0 on ZooKeeper (and Memory/RocksDB restart the version when a path is recreated). If our session expires after the `store.get` above but before this delete runs, and another worker then acquires the same lock path, the replacement also has version 0 and `delete(path, Optional.of(0))` removes it. The lock-id and `isCreatedBySelf` checks are done on the earlier read, not atomically with the delete. Oxia is fine because its version ids are unique per modification.
The window is small and this is much tighter than the unconditional delete it replaces, so I would not block on it. Two options: say in the comment that the conditional delete only narrows the race on version-resetting backends, or, if a backend-level fence is practical, tie the delete to the acquisition incarnation. The existing replacement tests (testLockCleanupPreservesReplacementInSameSession and friends) replace the node before the ownership read, so a deterministic test that replaces it between the read and the delete would show what is and is not covered.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -686,7 +698,70 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted while connecting metadata store", ie);
}
- heldLocks.remove(ledgerId);
+ if (l != null) {
+ heldLocks.remove(ledgerId, l);
+ }
+ }
+
+ private CompletableFuture<Void> releaseLock(long ledgerId, Lock lock) {
+ if (heldLocks.get(ledgerId) != lock) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ CompletableFuture<Void> existingRelease = lock.releaseFuture.compareAndExchange(null, result);
+ if (existingRelease != null) {
+ return existingRelease;
+ }
+
+ // A timed-out or interrupted wait does not cancel the delete. Reuse its outcome instead
+ // of sending another delete, which could target a lock acquired by a different manager.
+ MigrationPhase migrationPhase = getStoreMigrationPhase();
+ FutureUtil.supplySafely(() -> {
+ if (migrationPhase == MigrationPhase.PREPARATION || migrationPhase == MigrationPhase.COPYING) {
+ return CompletableFuture.<Optional<GetResult>>failedFuture(
+ new MetadataStoreException("Metadata migration is in progress; retry lock cleanup"));
+ }
+ return store.get(lock.getLockPath());
+ })
+ .thenCompose(existing -> {
+ if (getStoreMigrationPhase() != migrationPhase) {
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Metadata migration changed during lock cleanup; retry"));
+ }
+ if (existing.isEmpty() || !ownsLock(ledgerId, lock, existing.get())) {
+ if (migrationPhase == MigrationPhase.FAILED) {
Review Comment:
**[SUGGESTION] cleanup in the FAILED migration phase can leave the target-store copy of an owned lock behind**
The FAILED guard only applies when the source lock is missing or not ours. If the source copy is still ours (migration failed after PREPARATION already recreated the ephemeral node in the target), we skip the guard, delete the source copy and drop the entry from heldLocks. DualMetadataStore.delete in FAILED also removes the path from `localEphemeralPaths` and only touches the source store, so a later migration retry does not delete the target copy, and the target session keeps it alive, blocking rereplication for that ledger after cutover.
This is a general limitation of DualMetadataStore ephemerals rather than something this PR introduced, so a follow-up is fine, but since the comment above claims the failed-migration case is handled, it would be worth either deleting the owned copy from the target too (for example via a DualMetadataStore hook) or narrowing the comment. A case in testLockCleanupAfterMetadataMigration that runs cleanup while the phase is FAILED and the source lock is still present would pin it.
--
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] Preserve ownership during underreplication lock cleanup [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26794:
URL: https://github.com/apache/pulsar/pull/26794#discussion_r4215254372
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -686,7 +698,70 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted while connecting metadata store", ie);
}
- heldLocks.remove(ledgerId);
+ if (l != null) {
+ heldLocks.remove(ledgerId, l);
+ }
+ }
+
+ private CompletableFuture<Void> releaseLock(long ledgerId, Lock lock) {
+ if (heldLocks.get(ledgerId) != lock) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ CompletableFuture<Void> existingRelease = lock.releaseFuture.compareAndExchange(null, result);
+ if (existingRelease != null) {
+ return existingRelease;
+ }
+
+ // A timed-out or interrupted wait does not cancel the delete. Reuse its outcome instead
+ // of sending another delete, which could target a lock acquired by a different manager.
+ MigrationPhase migrationPhase = getStoreMigrationPhase();
+ FutureUtil.supplySafely(() -> {
+ if (migrationPhase == MigrationPhase.PREPARATION || migrationPhase == MigrationPhase.COPYING) {
+ return CompletableFuture.<Optional<GetResult>>failedFuture(
+ new MetadataStoreException("Metadata migration is in progress; retry lock cleanup"));
+ }
+ return store.get(lock.getLockPath());
+ })
+ .thenCompose(existing -> {
+ if (getStoreMigrationPhase() != migrationPhase) {
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Metadata migration changed during lock cleanup; retry"));
+ }
+ if (existing.isEmpty() || !ownsLock(ledgerId, lock, existing.get())) {
+ if (migrationPhase == MigrationPhase.FAILED) {
+ // A failed migration may still retain this acquisition in the inactive target.
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Lock ownership is uncertain after failed migration"));
+ }
+ return CompletableFuture.<Void>completedFuture(null);
+ }
+ return store.delete(lock.getLockPath(), Optional.of(existing.get().getStat().getVersion()));
Review Comment:
I've documented the remaining read-to-delete window and added a deterministic test that replaces the lock after the ownership read but before the actual delete. The ZooKeeper case expires the real source session, and the test covers all store implementations. Oxia rejects the stale version; ZooKeeper, Memory and RocksDB can reuse the version and still permit deletion of the replacement. This PR narrows that race and does not claim an atomic incarnation fence.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -686,7 +698,70 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted while connecting metadata store", ie);
}
- heldLocks.remove(ledgerId);
+ if (l != null) {
+ heldLocks.remove(ledgerId, l);
+ }
+ }
+
+ private CompletableFuture<Void> releaseLock(long ledgerId, Lock lock) {
+ if (heldLocks.get(ledgerId) != lock) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ CompletableFuture<Void> existingRelease = lock.releaseFuture.compareAndExchange(null, result);
+ if (existingRelease != null) {
+ return existingRelease;
+ }
+
+ // A timed-out or interrupted wait does not cancel the delete. Reuse its outcome instead
+ // of sending another delete, which could target a lock acquired by a different manager.
+ MigrationPhase migrationPhase = getStoreMigrationPhase();
+ FutureUtil.supplySafely(() -> {
+ if (migrationPhase == MigrationPhase.PREPARATION || migrationPhase == MigrationPhase.COPYING) {
+ return CompletableFuture.<Optional<GetResult>>failedFuture(
+ new MetadataStoreException("Metadata migration is in progress; retry lock cleanup"));
+ }
+ return store.get(lock.getLockPath());
+ })
+ .thenCompose(existing -> {
+ if (getStoreMigrationPhase() != migrationPhase) {
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Metadata migration changed during lock cleanup; retry"));
+ }
+ if (existing.isEmpty() || !ownsLock(ledgerId, lock, existing.get())) {
+ if (migrationPhase == MigrationPhase.FAILED) {
Review Comment:
I've added a DualMetadataStore hook to clean up the owned target copy before deleting the source lock. It tracks submitted copies per acquisition and checks target value and session ownership before deletion. Source-only acquisitions skip the target, so an unavailable target cannot block their release.
When the source lock has disappeared or been replaced, cleanup only handles the old target copy. Partial failures retain tracking for retry. Preparation also checks migration phase and source-session ownership to avoid copying a later or foreign acquisition under the wrong target session.
I kept this in the current PR to address the cleanup case raised here. The regressions cover release/close, target outages, source-session expiration, replacement ownership and migration retries through real backend paths.
--
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] Preserve ownership during underreplication lock cleanup [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26794:
URL: https://github.com/apache/pulsar/pull/26794#discussion_r4254888026
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -686,7 +698,70 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted while connecting metadata store", ie);
}
- heldLocks.remove(ledgerId);
+ if (l != null) {
+ heldLocks.remove(ledgerId, l);
+ }
+ }
+
+ private CompletableFuture<Void> releaseLock(long ledgerId, Lock lock) {
+ if (heldLocks.get(ledgerId) != lock) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ CompletableFuture<Void> existingRelease = lock.releaseFuture.compareAndExchange(null, result);
+ if (existingRelease != null) {
+ return existingRelease;
+ }
+
+ // A timed-out or interrupted wait does not cancel the delete. Reuse its outcome instead
+ // of sending another delete, which could target a lock acquired by a different manager.
+ MigrationPhase migrationPhase = getStoreMigrationPhase();
+ FutureUtil.supplySafely(() -> {
+ if (migrationPhase == MigrationPhase.PREPARATION || migrationPhase == MigrationPhase.COPYING) {
+ return CompletableFuture.<Optional<GetResult>>failedFuture(
+ new MetadataStoreException("Metadata migration is in progress; retry lock cleanup"));
+ }
+ return store.get(lock.getLockPath());
+ })
+ .thenCompose(existing -> {
+ if (getStoreMigrationPhase() != migrationPhase) {
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Metadata migration changed during lock cleanup; retry"));
+ }
+ if (existing.isEmpty() || !ownsLock(ledgerId, lock, existing.get())) {
+ if (migrationPhase == MigrationPhase.FAILED) {
+ // A failed migration may still retain this acquisition in the inactive target.
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Lock ownership is uncertain after failed migration"));
+ }
+ return CompletableFuture.<Void>completedFuture(null);
+ }
+ return store.delete(lock.getLockPath(), Optional.of(existing.get().getStat().getVersion()));
Review Comment:
Thanks, this is what I was after. The comment at :738-745 now states the limit, and testLockCleanupVersionFenceAcrossReplacement shows both outcomes by backend. I agree an atomic incarnation fence is out of scope here.
##########
pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java:
##########
@@ -686,7 +698,70 @@ public void releaseUnderreplicatedLedger(long ledgerId) throws ReplicationExcept
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted while connecting metadata store", ie);
}
- heldLocks.remove(ledgerId);
+ if (l != null) {
+ heldLocks.remove(ledgerId, l);
+ }
+ }
+
+ private CompletableFuture<Void> releaseLock(long ledgerId, Lock lock) {
+ if (heldLocks.get(ledgerId) != lock) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<Void> result = new CompletableFuture<>();
+ CompletableFuture<Void> existingRelease = lock.releaseFuture.compareAndExchange(null, result);
+ if (existingRelease != null) {
+ return existingRelease;
+ }
+
+ // A timed-out or interrupted wait does not cancel the delete. Reuse its outcome instead
+ // of sending another delete, which could target a lock acquired by a different manager.
+ MigrationPhase migrationPhase = getStoreMigrationPhase();
+ FutureUtil.supplySafely(() -> {
+ if (migrationPhase == MigrationPhase.PREPARATION || migrationPhase == MigrationPhase.COPYING) {
+ return CompletableFuture.<Optional<GetResult>>failedFuture(
+ new MetadataStoreException("Metadata migration is in progress; retry lock cleanup"));
+ }
+ return store.get(lock.getLockPath());
+ })
+ .thenCompose(existing -> {
+ if (getStoreMigrationPhase() != migrationPhase) {
+ return CompletableFuture.<Void>failedFuture(
+ new MetadataStoreException("Metadata migration changed during lock cleanup; retry"));
+ }
+ if (existing.isEmpty() || !ownsLock(ledgerId, lock, existing.get())) {
+ if (migrationPhase == MigrationPhase.FAILED) {
Review Comment:
Thanks, this covers the case I raised. The hook deletes the owned target copy before the source, and the source-only path stays releasable when the target is down. The tests exercise release, close, expiry, replacement and retry. Keeping it in this PR makes sense to me.
--
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] Preserve ownership during underreplication lock cleanup [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #26794:
URL: https://github.com/apache/pulsar/pull/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] Preserve ownership during underreplication lock cleanup [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on PR #26794:
URL: https://github.com/apache/pulsar/pull/26794#issuecomment-6146444528
https://github.com/apache/pulsar/actions/runs/37733581998/job/115137321819?pr=26794
@lhotari Like temporary download anomalies or dependency cache issues.
--
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