{"record":{"id":"efa705ca06748d3c","repo":"apache/flink","slug":"unrecognized-version-or-corrupt-state-version","errorCode":null,"errorMessage":"Unrecognized version or corrupt state: {version}","messagePattern":"Unrecognized version or corrupt state: (.+?)","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSinkCommittableSerializer.java","lineNumber":82,"sourceCode":"        DataOutputSerializer out = new DataOutputSerializer(256);\n        out.writeInt(MAGIC_NUMBER);\n        serializeV2(committable, out);\n        return out.getCopyOfBuffer();\n    }\n\n    @Override\n    public FileSinkCommittable deserialize(int version, byte[] serialized) throws IOException {\n        DataInputDeserializer in = new DataInputDeserializer(serialized);\n\n        switch (version) {\n            case 1:\n                validateMagicNumber(in);\n                return deserializeV1(in);\n            case 2:\n                validateMagicNumber(in);\n                return deserializeV2(in);\n            default:\n                throw new IOException(\"Unrecognized version or corrupt state: \" + version);\n        }\n    }\n\n    private void serializeV2(FileSinkCommittable committable, DataOutputView dataOutputView)\n            throws IOException {\n        dataOutputView.writeUTF(committable.getBucketId());\n        if (committable.hasPendingFile()) {\n            dataOutputView.writeBoolean(true);\n            SimpleVersionedSerialization.writeVersionAndSerialize(\n                    pendingFileSerializer, committable.getPendingFile(), dataOutputView);\n        } else {\n            dataOutputView.writeBoolean(false);\n        }\n\n        if (committable.hasInProgressFileToCleanup()) {\n            dataOutputView.writeBoolean(true);\n            SimpleVersionedSerialization.writeVersionAndSerialize(\n                    inProgressFileSerializer,","sourceCodeStart":64,"sourceCodeEnd":100,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSinkCommittableSerializer.java#L64-L100","documentation":"Thrown during deserialization of a FileSinkCommittable when the version number is not 1 or 2. The serializer supports two versions: v1 (legacy, pre-bucket-id) and v2 (current, with bucket id and compacted file cleanup). An unrecognized version indicates state incompatibility or corruption.","triggerScenarios":"Restoring a checkpoint or savepoint that was written by a different version of Flink with an incompatible committable serialization format; corrupted state data where the version byte was overwritten; manually tampered-with state.","commonSituations":"Upgrading or downgrading Flink across a version boundary that changed the committable serializer version; restoring from a savepoint produced by a different sink implementation; state backend corruption.","solutions":["Ensure the Flink version creating the checkpoint is compatible with the version restoring it.","If upgrading across major versions, follow the Flink upgrade and state compatibility guide.","If the state is genuinely corrupt or incompatible, start from a clean state without restoring from the savepoint.","Check the version number in the error message to determine the mismatch direction."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Check serializer version compatibility before restore\nFileSinkCommittableSerializer serializer = ...;\nint currentVersion = serializer.getVersion();\n// Ensure the checkpoint was written with version <= currentVersion\n// Version 1 = legacy format, Version 2 = current with bucketId + compacted cleanup","typeGuard":null,"tryCatchPattern":"try {\n    serializer.deserialize(version, data);\n} catch (IOException e) {\n    if (e.getMessage().startsWith(\"Unrecognized version or corrupt state\")) {\n        // version mismatch — cannot restore from this checkpoint\n        // start fresh or use a compatible savepoint\n    }\n    throw e;\n}","preventionTips":["Check Flink version compatibility before restoring checkpoints across versions.","Follow the Flink upgrade guide for state compatibility.","Keep track of which Flink version produced each savepoint."],"tags":["file-sink","serialization","state-restore","version-mismatch","checkpoint"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}