{"record":{"id":"c0212071a76db814","repo":"apache/seatunnel","slug":"packet-decode-error-c02120","errorCode":"PACKET_DECODE_ERROR","errorMessage":"Restore edge socket batch checkpoint state failed","messagePattern":"Restore edge socket batch checkpoint state failed","errorType":"error_code","errorClass":"EdgeSocketConnectorException","httpStatus":null,"severity":"error","filePath":"seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/state/EdgeSocketSourceState.java","lineNumber":286,"sourceCode":"\n            if (in.available() == 0) {\n                return new ArrayList<>();\n            }\n\n            int queuedSize = in.readInt();\n            EdgeSocketCompressionType[] compressionValues = EdgeSocketCompressionType.values();\n            List<EdgeSocketQueuedRecord> records = new ArrayList<>(queuedSize);\n            for (int i = 0; i < queuedSize; i++) {\n                long batchId = in.readLong();\n                EdgeSocketCompressionType compression = compressionValues[in.readInt()];\n                int payloadLen = in.readInt();\n                byte[] payload = new byte[payloadLen];\n                in.readFully(payload);\n                records.add(new EdgeSocketQueuedRecord(batchId, payload, compression));\n            }\n            return records;\n        } catch (IOException deserializeException) {\n            throw new EdgeSocketConnectorException(\n                    EdgeSocketConnectorErrorCode.PACKET_DECODE_ERROR,\n                    \"Restore edge socket batch checkpoint state failed\",\n                    deserializeException);\n        }\n    }\n\n    public void notifyCheckpointComplete(long checkpointId) {\n        Long completedWatermark = checkpointBatchWatermarks.remove(checkpointId);\n        if (completedWatermark == null) {\n            return;\n        }\n        if (completedWatermark > lastCommittedBatchId) {\n            lastCommittedBatchId = completedWatermark;\n        }\n        clearCommittedBatchState(lastCommittedBatchId);\n    }\n\n    public void notifyCheckpointAborted(long checkpointId) {","sourceCodeStart":268,"sourceCodeEnd":304,"githubUrl":"https://github.com/apache/seatunnel/blob/cf67b549a7a6c35fa0beb12d83c62892427ea919/seatunnel-connectors-v2/connector-edge-socket/src/main/java/org/apache/seatunnel/connectors/seatunnel/edgesocket/state/EdgeSocketSourceState.java#L268-L304","documentation":"Deserializing the checkpoint snapshot bytes in EdgeSocketSourceState failed with an IOException while reading queued records (batch ids, lengths, payloads). The connector wraps it as PACKET_DECODE_ERROR 'Restore edge socket batch checkpoint state failed', meaning the snapshot state blob is corrupt or written by an incompatible format.","triggerScenarios":"restoreState (public, called by records and watermark/ack restoration paths) reads the DataInput stream and encounters truncated data, wrong lengths, or an unexpected header — e.g. a snapshot from a different connector version with a changed serialization layout.","commonSituations":"See trigger scenarios.","solutions":["Restore from an earlier intact checkpoint/savepoint taken with the same connector version.","Verify the checkpoint file is complete and not truncated (compare sizes, checksums if available).","Align connector/engine versions between the job that took the snapshot and the job restoring it.","If the state is unrecoverable, restart the source without state and re-ingest from the edge producers."],"exampleFix":"// before: restoring a v2-format snapshot into a v1-serialized build\n./seatunnel.sh -s checkpoint-123\n// after: use a checkpoint from the matching connector version\n./seatunnel.sh -s checkpoint-120","handlingStrategy":"try-catch","validationCode":"// before restore, sanity-check state blob\nif (restoredState == null || restoredState.length < MIN_HEADER_BYTES) {\n    throw new IllegalStateException(\"checkpoint state missing or truncated\");\n}","typeGuard":null,"tryCatchPattern":"try {\n    reader.restoreState(state);\n} catch (EdgeSocketConnectorException e) {\n    log.error(\"checkpoint state unreadable (corrupt or version mismatch)\", e);\n    // fall back to an earlier checkpoint or restart without state\n}","preventionTips":["Restore snapshots with the same connector/engine version that wrote them.","Enable checkpoint file integrity checks and keep multiple recovery points.","Avoid manually copying checkpoint files across mismatched clusters.","Test savepoint/restore in staging after every connector upgrade."],"tags":["checkpoint","restore","serialization","corrupt-state"],"backgroundTag":"checkpoint-restore-failed","analyzedSha":"cf67b549a7a6c35fa0beb12d83c62892427ea919","analyzedAt":"2026-09-10T21:44:55.265Z","contentChangedAt":"2026-09-10T21:44:55.265Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}