{"record":{"id":"cdf014818ca1bf3a","repo":"apache/pulsar","slug":"timeout-during-reset-cursor","errorCode":null,"errorMessage":"Timeout during reset cursor","messagePattern":"Timeout during reset cursor","errorType":"exception","errorClass":"ManagedLedgerException","httpStatus":null,"severity":"error","filePath":"managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java","lineNumber":1791,"sourceCode":"                counter.countDown();\n            }\n\n            @Override\n            public void resetFailed(ManagedLedgerException exception, Object ctx) {\n                result.exception = exception;\n                counter.countDown();\n\n            }\n        });\n\n        if (!counter.await(ManagedLedgerImpl.AsyncOperationTimeoutSeconds, TimeUnit.SECONDS)) {\n            if (result.exception != null) {\n                log.warn()\n                        .attr(\"position\", newPos)\n                        .attr(\"error\", result.exception)\n                        .log(\"Reset cursor timed out\");\n            }\n            throw new ManagedLedgerException(\"Timeout during reset cursor\");\n        }\n\n        if (result.exception != null) {\n            throw result.exception;\n        }\n    }\n\n    @Override\n    public List<Entry> replayEntries(Set<? extends Position> positions)\n            throws InterruptedException, ManagedLedgerException {\n        final CountDownLatch counter = new CountDownLatch(1);\n        class Result {\n            ManagedLedgerException exception = null;\n            List<Entry> entries = null;\n        }\n\n        final Result result = new Result();\n","sourceCodeStart":1773,"sourceCodeEnd":1809,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java#L1773-L1809","documentation":"fromStream computes toSkip = headerLen - bytes already consumed and calls dis.skip(toSkip); if the underlying stream cannot skip that many bytes it throws EOFException(\"Header was too small\"). This means the InputStream contains fewer bytes than the header's declared headerLen, so the header cannot be fully consumed to reach the block payload. It is an end-of-data condition caused by truncated input or a malformed/inflated headerLen field.","triggerScenarios":"Calling fromStream with a stream carrying fewer than headerLen readable bytes: a truncated blob download, a header whose headerLen field was corrupted to a huge value, or a HEADER_MAX_SIZE-bounded stream where the declared header exceeds the available bytes.","commonSituations":"Interrupted or partial reads from object storage (network cut mid-read), offload blobs written by a crashed producer, corrupted headerLen bytes flipping to a large value, or reading a blob with a different format version whose header layout differs.","solutions":["Re-read the blob from storage — the input is truncated; delete and re-offload the ledger if the stored object is incomplete.","Validate the blob size on download (compare against stored object metadata / Content-Length) before parsing.","Check headerLen for sanity (e.g. < HEADER_MAX_SIZE) before skipping; an absurd value indicates header corruption.","Ensure reader and writer versions agree on the header format so headerLen is parsed from the right offset."],"exampleFix":"// before: no size check, truncated stream surfaces as EOF deep in parse\nStreamingDataBlockHeaderImpl.fromStream(truncatedStream);\n\n// after: verify declared header fits in available bytes first\nStreamingDataBlockHeaderImpl hdr = StreamingDataBlockHeaderImpl.of(headerLen, blockLen, ledgerId, firstEntryId);\nif (headerLen > availableBytes) {\n    throw new IOException(\"Truncated block: headerLen=\" + headerLen + \" available=\" + availableBytes);\n}","handlingStrategy":"validation","validationCode":"long declaredHeaderLen = readLongLE(bytes, offset); // once parsed\nif (bytesRemaining(stream) < declaredHeaderLen) {\n    throw new IOException(\"Truncated blob: headerLen=\" + declaredHeaderLen + \" > available bytes\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    return StreamingDataBlockHeaderImpl.fromStream(in);\n} catch (EOFException e) {\n    log.warn(\"Truncated offload data block header\", e);\n    return retryFromFreshDownload(); // re-open stream from storage, not the same stream\n}","preventionTips":["Compare downloaded blob size against stored object Content-Length/metadata before parsing.","Use fully-buffered reads for small blobs so a mid-read network failure fails cleanly.","Sanity-check headerLen against a max bound to catch corrupted header fields.","Monitor offload writes for completion; failed uploads leave truncated blobs."],"tags":["io","eof","truncation","offload"],"backgroundTag":"unexpected-eof","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"}