{"record":{"id":"b9c761d935d5ae15","repo":"apache/pulsar","slug":"failed-to-acuire-under-replicated-ledger","errorCode":null,"errorMessage":"Failed to acuire under-replicated ledger","messagePattern":"Failed to acuire under-replicated ledger","errorType":"exception","errorClass":"ReplicationException.UnavailableException","httpStatus":null,"severity":"error","filePath":"pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java","lineNumber":418,"sourceCode":"            store.put(path, newUrLedgerData, Optional.of(optRes.get().getStat().getVersion()))\n                    .thenRun(() -> {\n                        FutureUtils.complete(finalFuture, null);\n                    }).exceptionally(ex -> {\n                        FutureUtils.completeExceptionally(finalFuture, ex);\n                        return null;\n                    });\n        }).exceptionally(ex -> {\n            FutureUtils.completeExceptionally(finalFuture, ex);\n            return null;\n        });\n    }\n\n    @Override\n    public void acquireUnderreplicatedLedger(long ledgerId) throws ReplicationException {\n        try {\n            internalAcquireUnderreplicatedLedger(ledgerId);\n        } catch (ExecutionException | TimeoutException | InterruptedException e) {\n            throw new ReplicationException.UnavailableException(\"Failed to acuire under-replicated ledger\", e);\n        }\n    }\n\n    private void internalAcquireUnderreplicatedLedger(long ledgerId) throws ExecutionException,\n            InterruptedException, TimeoutException {\n        String lockPath = getUrLedgerLockPath(urLockPath, ledgerId);\n        store.put(lockPath, LOCK_DATA, Optional.of(-1L), EnumSet.of(CreateOption.Ephemeral))\n                .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);\n    }\n\n    @Override\n    public void markLedgerReplicated(long ledgerId) throws ReplicationException.UnavailableException {\n        log.debug().attr(\"ledgerId\", ledgerId).log(\"markLedgerReplicated\");\n        try {\n            Lock l = heldLocks.get(ledgerId);\n            if (l != null) {\n                store.delete(getUrLedgerPath(ledgerId), Optional.of(l.getLedgerNodeVersion()))\n                        .get(BLOCKING_CALL_TIMEOUT, MILLISECONDS);","sourceCodeStart":400,"sourceCodeEnd":436,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java#L400-L436","documentation":"Thrown by acquireUnderreplicatedLedger when internalAcquireUnderreplicatedLedger fails: the attempt to create the ephemeral lock node (store.put on the UR lock path with CreateOption.Ephemeral, blocked up to BLOCKING_CALL_TIMEOUT) throws ExecutionException, times out, or is interrupted. The rereplication worker did not obtain exclusive ownership of the ledger, so it must not proceed to replicate it.","triggerScenarios":"A BookKeeper Auditor/rereplication worker calls acquireUnderreplicatedLedger(ledgerId) when: the metadata store is unavailable (session expired, connection lost), the ephemeral lock creation exceeds BLOCKING_CALL_TIMEOUT, or the calling thread is interrupted while waiting for the put to complete.","commonSituations":"Multiple rereplication workers racing during a bookie failure while the metadata store is degraded; ZooKeeper failover in progress; broker shutdown interrupting the worker mid-acquire; network latency pushing lock creation past the blocking timeout.","solutions":["Retry acquiring the ledger after a backoff — another worker may legitimately hold or the outage may be transient.","Verify metadata store health (session, connectivity) before resuming rereplication work.","Check for a stale/leaked ephemeral lock path from a crashed worker; ephemeral nodes clear on session close, so confirm the dead worker's ZK session is gone.","Ensure worker threads are not interrupted during shutdown before acquisition completes; sequence shutdown after in-flight acquires."],"exampleFix":"// before: one failed acquire aborts the ledger processing\nurManager.acquireUnderreplicatedLedger(ledgerId);\nreplicate(ledgerId);\n// after: bounded retry around acquisition\nfor (int i = 0; i < 3; i++) {\n    try {\n        urManager.acquireUnderreplicatedLedger(ledgerId);\n        replicate(ledgerId);\n        return;\n    } catch (ReplicationException.UnavailableException e) {\n        Thread.sleep(1000L * (i + 1));\n    }\n}","handlingStrategy":"retry","validationCode":"// confirm the store is writable and reachable before attempting the lock\nmetadataStore.put(probePath, new byte[0], Optional.empty()).join();\n// optionally check an existing lock holder\nboolean locked = metadataStore.get(lockPath).join().isPresent();","typeGuard":"static boolean isAcquisitionFailure(ReplicationException e) {\n    return e instanceof ReplicationException.UnavailableException\n        && String.valueOf(e.getMessage()).contains(\"acuire\");\n}","tryCatchPattern":"try {\n    urManager.acquireUnderreplicatedLedger(ledgerId);\n} catch (ReplicationException.UnavailableException e) {\n    if (Thread.interrupted()) return; // shutdown\n    scheduleRetryWithBackoff(ledgerId); // transient: retry acquisition\n}","preventionTips":["Retry ledger acquisition with jittered backoff instead of dropping the ledger","Ensure crashed workers' ZK sessions expire so ephemeral locks release","Watch metadata store latency; scale the ensemble if puts routinely exceed the blocking timeout","Avoid shutting down workers while acquisition is in flight"],"tags":["distributed-lock","metadata-store","timeout","bookkeeper"],"backgroundTag":"lock-acquisition-failed","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}