{"record":{"id":"91b59ee5428536d0","repo":"apache/iceberg","slug":"unrecognized-version-or-corrupt-state-version-91b59e","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/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultSerializer.java","lineNumber":62,"sourceCode":"    view.writeInt(writeResult.specId());\n    byte[] result = WRITE_RESULT_SERIALIZER.serialize(writeResult.writeResult());\n    view.write(result);\n    return out.toByteArray();\n  }\n\n  @Override\n  public DynamicWriteResult deserialize(int version, byte[] serialized) throws IOException {\n    if (version == 1) {\n      DataInputDeserializer view = new DataInputDeserializer(serialized);\n      TableKey key = TableKey.deserializeFrom(view);\n      int specId = view.readInt();\n      byte[] resultBuf = new byte[view.available()];\n      view.read(resultBuf);\n      WriteResult writeResult = WRITE_RESULT_SERIALIZER.deserialize(version, resultBuf);\n      return new DynamicWriteResult(key, specId, writeResult);\n    }\n\n    throw new IOException(\"Unrecognized version or corrupt state: \" + version);\n  }\n}\n","sourceCodeStart":44,"sourceCodeEnd":65,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicWriteResultSerializer.java#L44-L65","documentation":"DynamicWriteResultSerializer.deserialize reads a version byte and delegates to the underlying WriteResult serializer only for versions it knows; any other version throws this IOException. Like the committable serializer, it protects against decoding write-result state produced by incompatible Iceberg versions.","triggerScenarios":"Restoring a Flink job whose operator state contains DynamicWriteResult bytes written with an unknown version byte (different/newer Iceberg sink version, or corrupted state). Called transitively via copy() when Flink duplicates state or manually via testUnsupportedVersion.","commonSituations":"Upgrading the Iceberg Flink sink between incompatible serializer versions and resuming from an old savepoint; mixed Iceberg jar versions across JobManager/TaskManagers; corrupted checkpoint bytes.","solutions":["Use the Iceberg version matching the savepoint's serialization version to restore","Rebuild job state without restore (start fresh; committed data remains in the table)","Check for classpath mixing of Iceberg Flink artifacts and align all nodes to one version"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Align iceberg-flink-runtime versions between the job that wrote the state and the restoring job.","typeGuard":null,"tryCatchPattern":"try {\n  result = serializer.deserialize(version, bytes);\n} catch (IOException e) {\n  // unknown version byte: restore with the matching Iceberg version or start without state\n}","preventionTips":["Keep JobManager and TaskManager Iceberg artifacts identical","Restore savepoints with the Iceberg version that wrote them","Exercise cross-version restore tests before upgrades"],"tags":["flink","serialization","versioning","checkpoint"],"backgroundTag":"unsupported-enum-value","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"}