{"record":{"id":"5b63e043426b2c3a","repo":"apache/flink","slug":"corrupt-data-unexpected-magic-number-5b63e0","errorCode":null,"errorMessage":"Corrupt data: Unexpected magic number.","messagePattern":"Corrupt data: Unexpected magic number\\.","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java","lineNumber":130,"sourceCode":"\n        return targetBytes;\n    }\n\n    @Override\n    public S3Recoverable deserialize(int version, byte[] serialized) throws IOException {\n        switch (version) {\n            case 1:\n                return deserializeV1(serialized);\n            default:\n                throw new IOException(\"Unrecognized version or corrupt state: \" + version);\n        }\n    }\n\n    private static S3Recoverable deserializeV1(byte[] serialized) throws IOException {\n        final ByteBuffer bb = ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN);\n\n        if (bb.getInt() != MAGIC_NUMBER) {\n            throw new IOException(\"Corrupt data: Unexpected magic number.\");\n        }\n\n        final byte[] keyBytes = new byte[bb.getInt()];\n        bb.get(keyBytes);\n\n        final byte[] uploadIdBytes = new byte[bb.getInt()];\n        bb.get(uploadIdBytes);\n\n        final int numParts = bb.getInt();\n        final ArrayList<PartETag> parts = new ArrayList<>(numParts);\n        for (int i = 0; i < numParts; i++) {\n            final int partNum = bb.getInt();\n            final byte[] buffer = new byte[bb.getInt()];\n            bb.get(buffer);\n            parts.add(new PartETag(partNum, new String(buffer, CHARSET)));\n        }\n\n        final long numBytes = bb.getLong();","sourceCodeStart":112,"sourceCodeEnd":148,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java#L112-L148","documentation":"Thrown by S3RecoverableSerializer.deserializeV1 when deserializing the persisted state of an in-progress S3 recoverable write (a CommitRecoverable snapshot). The serialized payload starts with a fixed MAGIC_NUMBER int (little-endian); if the first 4 bytes do not match, the bytes are not a valid S3Recoverable payload. This almost always means the state/savepoint data is truncated, corrupted, or was written by a different serializer.","triggerScenarios":"Calling S3RecoverableSerializer.deserialize() (directly or via RecoverableWriter.recover/resume/commit after checkpoint restore) with a byte[] that is not a version-1 S3Recoverable: truncated state, garbage bytes, little-endian/big-endian mismatch, or a payload produced by a different RecoverableSerializer (e.g. a different filesystem's writer).","commonSituations":"Restoring a job from a savepoint or checkpoint where the persisted recoverable-serializer state was corrupted (partial write to state backend), upgrading Flink across versions that changed the serializer format, or handing a recoverable from one filesystem implementation (e.g. HadoopS3) to the wrong writer. Also happens when operator state was manually edited or the state backend lost data.","solutions":["Verify the byte[] passed to deserialize came from S3RecoverableSerializer.serialize() of the same Flink version; do not mix recoverables across filesystem implementations.","Check the preceding version byte: the switch only accepts version 1, so payloads serialized by any other version fail earlier with 'Unrecognized version or corrupt state' — confirm you are on a matching Flink version on both write and restore.","If state corruption is suspected, verify checkpoint/savepoint integrity (sizes, checksums, storage durability) and restore from the last known-good checkpoint instead.","As a last resort, discard the in-progress recoverable data and rewrite the output (losing exactly-once append for that file), then re-run."],"exampleFix":"// before\nS3Recoverable rec = (S3Recoverable) S3RecoverableSerializer.INSTANCE.deserialize(corruptBytes);\n\n// after\nCommitRecoverable rec = S3RecoverableSerializer.INSTANCE.deserialize(bytes);\n// bytes must originate from S3RecoverableSerializer.serialize() of the same\n// Flink version and the same S3 RecoverableWriter; validate length first:\nif (bytes == null || bytes.length < 5) {\n    throw new IOException(\"State too short to be an S3Recoverable payload\");\n}","handlingStrategy":"validation","validationCode":"// before deserializing recoverable state\nprivate static final int MAGIC = S3RecoverableSerializer.MAGIC_NUMBER; // or inline constant\nstatic boolean looksLikeS3Recoverable(byte[] bytes) {\n    if (bytes == null || bytes.length < 5) return false; // version byte + 4 magic bytes\n    ByteBuffer bb = ByteBuffer.wrap(bytes, 1, 4).order(ByteOrder.LITTLE_ENDIAN);\n    return bb.getInt() == MAGIC;\n}","typeGuard":null,"tryCatchPattern":"try {\n    S3Recoverable rec = (S3Recoverable) S3RecoverableSerializer.INSTANCE.deserialize(bytes);\n} catch (IOException e) {\n    // treat as unrecoverable state: log checkpoint/savepoint id, do not resume,\n    // fall back to full rewrite of the output file\n}","preventionTips":["Always serialize and deserialize recoverables with S3RecoverableSerializer from the same Flink version on both sides.","Never hand recoverables between different filesystem implementations.","Monitor checkpoint storage durability; partial state writes surface here as magic-number failures."],"tags":["s3","serialization","state-restore","corruption"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}