{"record":{"id":"d29b24f3f7bfc138","repo":"apache/iceberg","slug":"unknown-version-version-d29b24","errorCode":null,"errorMessage":"Unknown version: {version}","messagePattern":"Unknown version: (.+?)","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/IcebergEnumeratorStateSerializer.java","lineNumber":69,"sourceCode":"  @Override\n  public int getVersion() {\n    return VERSION;\n  }\n\n  @Override\n  public byte[] serialize(IcebergEnumeratorState enumState) throws IOException {\n    return serializeV2(enumState);\n  }\n\n  @Override\n  public IcebergEnumeratorState deserialize(int version, byte[] serialized) throws IOException {\n    switch (version) {\n      case 1:\n        return deserializeV1(serialized);\n      case 2:\n        return deserializeV2(serialized);\n      default:\n        throw new IOException(\"Unknown version: \" + version);\n    }\n  }\n\n  @VisibleForTesting\n  byte[] serializeV1(IcebergEnumeratorState enumState) throws IOException {\n    DataOutputSerializer out = SERIALIZER_CACHE.get();\n    serializeEnumeratorPosition(out, enumState.lastEnumeratedPosition(), positionSerializer);\n    serializePendingSplits(out, enumState.pendingSplits(), splitSerializer);\n    byte[] result = out.getCopyOfBuffer();\n    out.clear();\n    return result;\n  }\n\n  @VisibleForTesting\n  IcebergEnumeratorState deserializeV1(byte[] serialized) throws IOException {\n    DataInputDeserializer in = new DataInputDeserializer(serialized);\n    IcebergEnumeratorPosition enumeratorPosition =\n        deserializeEnumeratorPosition(in, positionSerializer);","sourceCodeStart":51,"sourceCodeEnd":87,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/IcebergEnumeratorStateSerializer.java#L51-L87","documentation":"IcebergEnumeratorStateSerializer.deserialize throws IOException when the enumerator-state blob's version byte is neither 1 nor 2, i.e. state written by an incompatible connector version. It protects readers from decoding state layouts they don't understand.","triggerScenarios":"Restoring from a checkpoint/savepoint whose IcebergEnumeratorState bytes were serialized with a version outside {1,2}, typically by a newer connector release.","commonSituations":"Rolling upgrade where the job manager restored state produced by a newer flink-runtime jar; downgrading the connector; cross-version state migration between Flink 1.x streams.","solutions":["Restart the restore using the connector version that matches the state's writer version (same or newer).","Start fresh (no savepoint) and let the enumerator re-enumerate from the table if state migration isn't feasible.","Use Flink state-processor-api to inspect/rewrite the enumerator state to a supported version.","Align all jars in the deployment so writer and reader versions can never diverge."],"exampleFix":"// before\nenv.execute with connector 1.8.x restoring state written by 1.10.x\n// after\nupgrade job to iceberg-flink-runtime 1.10.x before restoring the savepoint","handlingStrategy":"try-catch","validationCode":"// ensure deployment jars match before resuming\nPreconditions.checkState(\n    IcebergBuild.version().equals(stateWritingIcebergVersion),\n    \"Connector version mismatch with checkpoint state\");","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Roll out connector upgrades with a savepoint taken and restored by the same version first.","Avoid mixing connector versions within one Flink cluster.","Document the state version bump in upgrade runbooks."],"tags":["flink","serialization","state","version-mismatch"],"backgroundTag":"incompatible-source-type","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}