{"record":{"id":"40b6332da951ec4d","repo":"apache/iceberg","slug":"unrecognized-version-or-corrupt-state-version-40b633","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/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java","lineNumber":70,"sourceCode":"    view.writeInt(numManifests);\n    for (int i = 0; i < numManifests; i++) {\n      byte[] manifest = committable.manifests()[i];\n      view.writeInt(manifest.length);\n      view.write(manifest);\n    }\n\n    return out.toByteArray();\n  }\n\n  @Override\n  public DynamicCommittable deserialize(int version, byte[] serialized) throws IOException {\n    if (version == VERSION_1) {\n      return deserializeV1(serialized);\n    } else if (version == VERSION_2) {\n      return deserializeV2(serialized);\n    }\n\n    throw new IOException(\"Unrecognized version or corrupt state: \" + version);\n  }\n\n  private DynamicCommittable deserializeV1(byte[] serialized) throws IOException {\n    DataInputDeserializer view = new DataInputDeserializer(serialized);\n    WriteTarget key = WriteTarget.deserializeFrom(view);\n    String jobId = view.readUTF();\n    String operatorId = view.readUTF();\n    long checkpointId = view.readLong();\n    int manifestLen = view.readInt();\n    byte[] manifestBuf = new byte[manifestLen];\n    view.read(manifestBuf);\n    return new DynamicCommittable(\n        new TableKey(key.tableName(), key.branch()),\n        new byte[][] {manifestBuf},\n        jobId,\n        operatorId,\n        checkpointId);\n  }","sourceCodeStart":52,"sourceCodeEnd":88,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommittableSerializer.java#L52-L88","documentation":"DynamicCommittableSerializer.deserialize dispatches on a version byte (VERSION_1/VERSION_2) and throws IOException when the byte matches neither, meaning the serialized committable came from an incompatible serializer version or the bytes are corrupt.","triggerScenarios":"Deserializing DynamicCommittables from a checkpoint/savepoint written by an Iceberg version with a different serializer version byte; corrupted committable bytes in operator state.","commonSituations":"Upgrading iceberg-flink runtime across a restore of in-flight dynamic-sink committables; state corruption in checkpoint storage.","solutions":["Restore with the Iceberg version that wrote the checkpoint, finish the pending commit, then upgrade.","Start fresh (no old savepoint) so committables are produced in the current version.","Validate checkpoint integrity and fall back to an earlier checkpoint if corruption is suspected."],"exampleFix":null,"handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  return serializer.deserialize(bytes);\n} catch (IOException e) {\n  if (e.getMessage().startsWith(\"Unrecognized version or corrupt state\")) {\n    // restart without the stale savepoint / use matching Iceberg version\n  }\n  throw e;\n}","preventionTips":["Stop-with-savepoint and drain before upgrading Iceberg.","Keep the dynamic-sink serializer versions consistent across upgrades.","Use checkpoint checksums to detect corruption early."],"tags":["flink","serialization","dynamic-sink","version-skew"],"backgroundTag":"unexpected-response-shape","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"}