{"record":{"id":"26b3a75ab17b39b5","repo":"apache/pulsar","slug":"error-parsing-proto-message","errorCode":null,"errorMessage":"Error parsing proto message","messagePattern":"Error parsing proto message","errorType":"exception","errorClass":"ReplicationException.UnavailableException","httpStatus":null,"severity":"error","filePath":"pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java","lineNumber":321,"sourceCode":"            byte[] data = optRes.get().getValue();\n\n            UnderreplicatedLedgerFormat underreplicatedLedgerFormat = new UnderreplicatedLedgerFormat();\n\n            underreplicatedLedgerFormat.parseFromTextFormat(data);\n            PulsarUnderreplicatedLedger underreplicatedLedger = new PulsarUnderreplicatedLedger(ledgerId);\n            List<String> replicaList = underreplicatedLedgerFormat.getReplicasList();\n            long ctime = (underreplicatedLedgerFormat.hasCtime() ? underreplicatedLedgerFormat.getCtime()\n                    : UnderreplicatedLedger.UNASSIGNED_CTIME);\n            underreplicatedLedger.setCtime(ctime);\n            underreplicatedLedger.setReplicaList(replicaList);\n            return underreplicatedLedger;\n        } catch (ExecutionException | TimeoutException ee) {\n            throw new ReplicationException.UnavailableException(\"Error contacting with metadata store\", ee);\n        } catch (InterruptedException ie) {\n            Thread.currentThread().interrupt();\n            throw new ReplicationException.UnavailableException(\"Interrupted while connecting metadata store\", ie);\n        } catch (RuntimeException pe) {\n            throw new ReplicationException.UnavailableException(\"Error parsing proto message\", pe);\n        }\n    }\n\n    @Override\n    public CompletableFuture<Void> markLedgerUnderreplicatedAsync(long ledgerId, Collection<String> missingReplicas) {\n        log.debug().attr(\"ledgerId\", ledgerId).attr(\"missingReplicas\", missingReplicas)\n                .log(\"markLedgerUnderreplicated\");\n        final String path = getUrLedgerPath(ledgerId);\n        final CompletableFuture<Void> createFuture = new CompletableFuture<>();\n        tryMarkLedgerUnderreplicatedAsync(path, missingReplicas, createFuture);\n        return createFuture;\n    }\n\n    private void tryMarkLedgerUnderreplicatedAsync(final String path,\n                                                   final Collection<String> missingReplicas,\n                                                   final CompletableFuture<Void> finalFuture) {\n        final UnderreplicatedLedgerFormat builder = new UnderreplicatedLedgerFormat();\n        if (conf.getStoreSystemTimeAsLedgerUnderreplicatedMarkTime()) {","sourceCodeStart":303,"sourceCodeEnd":339,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/bookkeeper/PulsarLedgerUnderreplicationManager.java#L303-L339","documentation":"Thrown when the raw bytes fetched from the metadata store cannot be parsed into the UnderreplicatedLedgerFormat protobuf text format (parseFromTextFormat throws a RuntimeException). This indicates the under-replicated ledger znode/record at the ledger's metadata path is corrupt or was written by an incompatible version, not a connectivity problem.","triggerScenarios":"Calling getLedgerUnreplicationInfo for a ledger whose stored under-replicated marker contains non-text, truncated, or schema-incompatible data — e.g. the node was written by an older/newer Pulsar with a different proto field layout, or manually edited/corrupted znode data.","commonSituations":"After a partial metadata-store migration or restore from backup with inconsistent data; manual znode edits via zkCli; version skew between brokers and BookKeeper audit components during a rolling upgrade; disk-level corruption in ZooKeeper transaction logs.","solutions":["Inspect the raw data at the ledger path (e.g. 'get /ledgers/underreplicated/...') and identify the malformed content.","Delete the corrupt under-replicated ledger node so the Auditor can re-mark it if the ledger is still under-replicated.","Ensure all brokers and the auditor run compatible Pulsar/BookKeeper versions to avoid proto format mismatch.","If corruption is widespread, restore the underreplicated-ledger subtree from a consistent backup or trigger a full ledger re-check."],"exampleFix":"// before: repeatedly failing on corrupt node\nUnderreplicatedLedger l = urManager.getLedgerUnreplicationInfo(ledgerId); // throws every time\n// after: clean the corrupt marker and let re-marking recreate it\ntry {\n    return urManager.getLedgerUnreplicationInfo(ledgerId);\n} catch (ReplicationException.UnavailableException e) {\n    log.warn(\"corrupt UR data for {}\", ledgerId, e);\n    metadataStore.delete(urLedgerPath(ledgerId), Optional.empty()).join();\n    return null;\n}","handlingStrategy":"validation","validationCode":"// verify the stored payload parses before relying on it\nbyte[] data = metadataStore.get(urLedgerPath).join().get().getValue();\ntry {\n    new UnderreplicatedLedgerFormat().parseFromTextFormat(data);\n} catch (RuntimeException e) {\n    // corrupt: quarantine or delete the node before further reads\n}","typeGuard":"static boolean isParseFailure(ReplicationException.UnavailableException e) {\n    Throwable c = e.getCause();\n    return c instanceof RuntimeException && !(c instanceof MetadataStoreException);\n}","tryCatchPattern":"try {\n    return urManager.getLedgerUnreplicationInfo(ledgerId);\n} catch (ReplicationException.UnavailableException e) {\n    if (e.getCause() instanceof RuntimeException) {\n        log.warn(\"corrupt UR marker for {}, deleting node\", ledgerId, e);\n        metadataStore.delete(urLedgerPath, Optional.empty()).join();\n        return null;\n    }\n    throw e;\n}","preventionTips":["Never hand-edit under-replicated ledger nodes via zkCli","Keep all brokers/auditors on compatible Pulsar versions during rolling upgrades","Validate metadata backups before restoring them into a live cluster","Monitor ZooKeeper data integrity and enable snapshot verification"],"tags":["corruption","protobuf","parsing","data-format"],"backgroundTag":"corrupt-protobuf-data","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"}