{"record":{"id":"1c1773b50874a091","repo":"apache/iceberg","slug":"unrecognized-version-or-corrupt-state-version","errorCode":null,"errorMessage":"Unrecognized version or corrupt state: <version>","messagePattern":"Unrecognized version or corrupt state: <version>","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/MonitorSource.java","lineNumber":202,"sourceCode":"          iterator.getClass());\n\n      TableChangeIterator tableChangeIterator = (TableChangeIterator) iterator;\n      DataOutputSerializer out = new DataOutputSerializer(8);\n      long toStore =\n          tableChangeIterator.lastSnapshotId != null ? tableChangeIterator.lastSnapshotId : -1L;\n      out.writeLong(toStore);\n      return out.getCopyOfBuffer();\n    }\n\n    @Override\n    public TableChangeIterator deserialize(int version, byte[] serialized) throws IOException {\n      if (version == CURRENT_VERSION) {\n        DataInputDeserializer in = new DataInputDeserializer(serialized);\n        long fromStore = in.readLong();\n        return new TableChangeIterator(\n            tableLoader, fromStore != -1 ? fromStore : null, maxReadBack);\n      } else {\n        throw new IOException(\"Unrecognized version or corrupt state: \" + version);\n      }\n    }\n  }\n}\n","sourceCodeStart":184,"sourceCodeEnd":207,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/MonitorSource.java#L184-L207","documentation":"MonitorSource's serializer writes a version byte/long alongside checkpoint state. On restore, if the stored state version is not the current version (e.g. state written by a different Iceberg/Flink release or truncated state), deserialization fails with an IOException indicating unrecognized version or corrupt state.","triggerScenarios":"Restoring a Flink job from a savepoint/checkpoint whose operator state for the Iceberg MonitorSource was written by an incompatible Iceberg version, or reading truncated/edited state bytes.","commonSituations":"Upgrading the Iceberg Flink runtime and resuming from an old savepoint; migrating checkpoints across Iceberg major versions; corrupted checkpoint storage (HDFS/S3 truncation).","solutions":["Use the same (or a compatible) Iceberg version that wrote the savepoint/checkpoint to restore state, then upgrade with state migration if supported.","Start a new job without state (drop the savepoint) and let the MonitorSource rebuild its TableChange state.","Verify checkpoint files are complete and uncorrupted; re-take the savepoint from a healthy checkpoint."],"exampleFix":"// before\nbin/flink run -s old-savepoint-from-iceberg-1.4.jar ...\n// after\n// upgrade the job graph AND iceberg-flink-runtime to the version that wrote the savepoint first\nbin/flink run -s savepoint --jarfile matching-iceberg-version-runtime.jar ...","handlingStrategy":"fallback","validationCode":"// Before restoring, check the Iceberg version that wrote the savepoint matches the runtime version","typeGuard":null,"tryCatchPattern":"try { restoreFromSavepoint(savepoint); } catch (IOException e) { if (e.getMessage().contains(\"Unrecognized version\")) { startWithoutState(); } else { throw e; } }","preventionTips":["Pin iceberg-flink-runtime version across savepoints and restores","Upgrade in two steps: restore with old version, take new savepoint, then upgrade runtime","Test savepoint compatibility in staging before production upgrades"],"tags":["flink","checkpoint","serialization"],"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"}