{"record":{"id":"6d3434e5f9b0f4d1","repo":"apache/flink","slug":"the-bytes-are-serialized-with-version-d-while-th","errorCode":null,"errorMessage":"The bytes are serialized with version %d, while this deserializer only supports version up to %d","messagePattern":"The bytes are serialized with version (.+?), while this deserializer only supports version up to (.+?)","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"critical","filePath":"flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceEnumeratorStateSerializer.java","lineNumber":61,"sourceCode":"    public byte[] serialize(HybridSourceEnumeratorState enumState) throws IOException {\n        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();\n                DataOutputStream out = new DataOutputStream(baos)) {\n            out.writeInt(enumState.getCurrentSourceIndex());\n            out.writeInt(enumState.getWrappedStateSerializerVersion());\n            out.writeInt(enumState.getWrappedState().length);\n            out.write(enumState.getWrappedState());\n            out.flush();\n            return baos.toByteArray();\n        }\n    }\n\n    @Override\n    public HybridSourceEnumeratorState deserialize(int version, byte[] serialized)\n            throws IOException {\n        if (version == 0) {\n            return deserializeV0(serialized);\n        }\n        throw new IOException(\n                String.format(\n                        \"The bytes are serialized with version %d, \"\n                                + \"while this deserializer only supports version up to %d\",\n                        version, CURRENT_VERSION));\n    }\n\n    private HybridSourceEnumeratorState deserializeV0(byte[] serialized) throws IOException {\n        try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);\n                DataInputStream in = new DataInputStream(bais)) {\n            int sourceIndex = in.readInt();\n            int nestedVersion = in.readInt();\n            int length = in.readInt();\n            byte[] nestedBytes = new byte[length];\n            in.readFully(nestedBytes);\n            return new HybridSourceEnumeratorState(sourceIndex, nestedBytes, nestedVersion);\n        }\n    }\n}","sourceCodeStart":43,"sourceCodeEnd":79,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/hybrid/HybridSourceEnumeratorStateSerializer.java#L43-L79","documentation":"Thrown as an IOException by HybridSourceEnumeratorStateSerializer.deserialize when the serialized state version is not 0 (CURRENT_VERSION). The serializer only supports version 0; any other version indicates the state was produced by a future or incompatible version of HybridSourceEnumeratorStateSerializer. This is a forward-compatibility guard: the deserializer refuses to process bytes it cannot safely interpret.","triggerScenarios":"Restoring a HybridSource enumerator state from a checkpoint/savepoint produced by a newer Flink version that uses a higher serializer version (version > 0).","commonSituations":"Upgrading Flink to a version that bumped HybridSourceEnumeratorStateSerializer.getVersion(), then trying to restore an old client/deserializer against the newer checkpoint; downgrade scenario where a newer checkpoint is loaded by an older Flink version.","solutions":["Upgrade the Flink client and cluster to the version that produced the checkpoint (matching or newer serializer version).","If downgrading is required, start the job from scratch without restoring the HybridSource enumerator state.","Never downgrade across versions that change the HybridSource state serializer format.","Check the Flink release notes for HybridSource state serializer version changes before upgrading."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Check serializer version compatibility before deserializing\nint stateVersion = HybridSourceEnumeratorStateSerializer.INSTANCE.getVersion();\nif (checkpointVersion > stateVersion) {\n    throw new IllegalStateException(\n        \"Cannot restore: checkpoint was serialized with version \" + checkpointVersion\n        + \" but this serializer only supports up to version \" + stateVersion);\n}","typeGuard":null,"tryCatchPattern":"try {\n    HybridSourceEnumeratorState state = serializer.deserialize(version, bytes);\n} catch (IOException e) {\n    if (e.getMessage().contains(\"only supports version up to\")) {\n        // version mismatch — upgrade Flink or start fresh\n        log.error(\"State version mismatch: {}\", e.getMessage());\n    }\n}","preventionTips":["Never downgrade Flink across versions that change the HybridSource state serializer.","Check the serializer version in release notes before upgrading.","Take a fresh checkpoint after upgrading if state format changed."],"tags":["hybrid-source","state-serialization","version-mismatch","checkpoint-restore"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}