{"record":{"id":"f202fccc23b7e6c1","repo":"apache/flink","slug":"invalid-version-d","errorCode":null,"errorMessage":"Invalid version %d","messagePattern":"Invalid version (.+?)","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitSerializer.java","lineNumber":58,"sourceCode":"    public byte[] serialize(HybridSourceSplit split) throws IOException {\n        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();\n                DataOutputStream out = new DataOutputStream(baos)) {\n            out.writeInt(split.sourceIndex());\n            out.writeUTF(split.splitId());\n            out.writeInt(split.wrappedSplitSerializerVersion());\n            out.writeInt(split.wrappedSplitBytes().length);\n            out.write(split.wrappedSplitBytes());\n            out.flush();\n            return baos.toByteArray();\n        }\n    }\n\n    @Override\n    public HybridSourceSplit deserialize(int version, byte[] serialized) throws IOException {\n        if (version == 0) {\n            return deserializeV0(serialized);\n        }\n        throw new IOException(String.format(\"Invalid version %d\", version));\n    }\n\n    private HybridSourceSplit deserializeV0(byte[] serialized) throws IOException {\n        try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);\n                DataInputStream in = new DataInputStream(bais)) {\n            int sourceIndex = in.readInt();\n            String splitId = in.readUTF();\n            int nestedVersion = in.readInt();\n            int length = in.readInt();\n            byte[] splitBytes = new byte[length];\n            in.readFully(splitBytes);\n            return new HybridSourceSplit(sourceIndex, splitBytes, nestedVersion, splitId);\n        }\n    }\n}\n","sourceCodeStart":40,"sourceCodeEnd":74,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceSplitSerializer.java#L40-L74","documentation":"HybridSourceSplitSerializer#deserialize only knows how to read version 0 of the serialized split format. Any other version number (passed by the framework from SimpleVersionedSerializer) is rejected with an IOException. This guards against reading a split that was serialized by a newer or older incompatible serializer revision.","triggerScenarios":"Restoring a checkpoint/savepoint whose HybridSourceSplit bytes were written with a serializer version other than 0; a custom serializer wrapper that bumps the outer version without handling it; downgrading Flink after splits were persisted under a newer format.","commonSituations":"Downgrading Flink versions where the split format version changed; mixing Flink versions in a recovery path; manual corruption of checkpoint metadata.","solutions":["Check the JobManager logs for the exact version number reported and compare against the current HybridSourceSplitSerializer constant (0).","If restoring across an incompatible Flink version, start a fresh job without the savepoint because the split bytes cannot be decoded.","If you maintain a fork that changed the version, add a deserializeV<n> branch mirroring the existing deserializeV0 pattern.","Ensure no custom code is calling serialize with a version other than the one returned by getVersion()."],"exampleFix":"// before\n@Override\npublic HybridSourceSplit deserialize(int version, byte[] serialized) throws IOException {\n    if (version == 0) return deserializeV0(serialized);\n    throw new IOException(String.format(\"Invalid version %d\", version));\n}\n// after: handle a new version explicitly\n@Override\npublic HybridSourceSplit deserialize(int version, byte[] serialized) throws IOException {\n    switch (version) {\n        case 0: return deserializeV0(serialized);\n        case 1: return deserializeV1(serialized);\n        default: throw new IOException(String.format(\"Invalid version %d\", version));\n    }\n}","handlingStrategy":"validation","validationCode":"int expected = new HybridSourceSplitSerializer().getVersion(); // 0\nif (version != expected) {\n    throw new IOException(\"Cannot restore HybridSourceSplit: version \" + version + \" != \" + expected);\n}","typeGuard":null,"tryCatchPattern":"try {\n    return serializer.deserialize(version, bytes);\n} catch (IOException e) {\n    if (e.getMessage().startsWith(\"Invalid version\")) {\n        // incompatible checkpoint; cannot recover this split\n        throw new FlinkRuntimeException(\"Incompatible HybridSourceSplit version: \" + version, e);\n    }\n    throw e;\n}","preventionTips":["Do not downgrade Flink past a split-format change without discarding checkpoints.","If forking, always add a deserializeV<n> branch when bumping getVersion().","Store the Flink version alongside checkpoints for restore-time diagnostics."],"tags":["serialization","checkpoint-restore","hybrid-source","version-mismatch","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}