{"record":{"id":"fc2dc818c7ad3711","repo":"apache/flink","slug":"unrecognized-version-or-corrupt-state","errorCode":null,"errorMessage":"Unrecognized version or corrupt state: {}","messagePattern":"Unrecognized version or corrupt state: (.+?)","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/core/fs/local/LocalRecoverableSerializer.java","lineNumber":72,"sourceCode":"\n        ByteBuffer bb = ByteBuffer.wrap(targetBytes).order(ByteOrder.LITTLE_ENDIAN);\n        bb.putInt(MAGIC_NUMBER);\n        bb.putLong(obj.offset());\n        bb.putInt(targetFileBytes.length);\n        bb.putInt(tempFileBytes.length);\n        bb.put(targetFileBytes);\n        bb.put(tempFileBytes);\n\n        return targetBytes;\n    }\n\n    @Override\n    public LocalRecoverable 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 LocalRecoverable 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 long offset = bb.getLong();\n        final byte[] targetFileBytes = new byte[bb.getInt()];\n        final byte[] tempFileBytes = new byte[bb.getInt()];\n        bb.get(targetFileBytes);\n        bb.get(tempFileBytes);\n\n        final String targetPath = new String(targetFileBytes, CHARSET);\n        final String tempPath = new String(tempFileBytes, CHARSET);","sourceCodeStart":54,"sourceCodeEnd":90,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/core/fs/local/LocalRecoverableSerializer.java#L54-L90","documentation":"Thrown by LocalRecoverableSerializer.deserialize when the supplied version integer does not match any known case in its switch statement (currently only version 1 is recognized). LocalRecoverableSerializer is the SimpleVersionedSerializer that persists LocalFileSystem file-sink recovery metadata (target file, temp file, offset). The version comes from SimpleVersionedSerialization framing, so a mismatch means the bytes were produced by a newer, older, or incompatible serializer implementation.","triggerScenarios":"Calling LocalRecoverableSerializer.deserialize(version, bytes) with a version other than 1; data produced by SimpleVersionedSerialization.readVersionAndDeSerialize where the embedded version field does not equal 1; restoring a checkpoint or savepoint whose local file-sink recovery metadata was serialized by a future or alternate serializer version.","commonSituations":"Restoring a job from a savepoint/checkpoint written by a different Flink version that changed the LocalRecoverable format; corrupted or truncated recovery metadata bytes where the version field reads as garbage; using a custom SimpleVersionedSerializer that delegates to LocalRecoverableSerializer with mismatched version wiring.","solutions":["Check the Flink version that wrote the checkpoint/savepoint against the version restoring it; LocalRecoverableSerializer has only ever had version 1, so a mismatch usually means the bytes are not actually a LocalRecoverable payload.","Verify the byte array passed to deserialize was produced by SimpleVersionedSerialization.writeVersionAndSerialize and not corrupted in transit or storage.","If migrating across Flink versions, consult the state compatibility / migration notes for the file sink and re-create the savepoint if the format is incompatible.","Add a guard that logs the version and byte length before calling deserialize to confirm the payload is well-formed."],"exampleFix":"// before\nLocalRecoverable r = LocalRecoverableSerializer.INSTANCE.deserialize(version, bytes);\n\n// after — guard against unknown versions\nif (version != LocalRecoverableSerializer.INSTANCE.getVersion()) {\n    throw new IOException(\"Unsupported LocalRecoverable version \" + version\n        + \"; expected \" + LocalRecoverableSerializer.INSTANCE.getVersion());\n}\nLocalRecoverable r = LocalRecoverableSerializer.INSTANCE.deserialize(version, bytes);","handlingStrategy":"validation","validationCode":"if (version != LocalRecoverableSerializer.INSTANCE.getVersion()) {\n    throw new IOException(\"Unsupported LocalRecoverable version \" + version);\n}","typeGuard":null,"tryCatchPattern":"try {\n    LocalRecoverable r = LocalRecoverableSerializer.INSTANCE.deserialize(version, bytes);\n} catch (IOException e) {\n    if (e.getMessage().contains(\"Unrecognized version\")) {\n        // handle version mismatch: log, re-create state, or re-throw\n    }\n    throw e;\n}","preventionTips":["Always read recovery metadata with SimpleVersionedSerialization.readVersionAndDeSerialize so version framing is handled consistently.","Do not mix Flink versions across write and restore of local file-sink recovery state.","Log the version and byte length before deserializing to aid diagnosis."],"tags":["serialization","local-filesystem","checkpoint-restore","version-mismatch"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}