{"record":{"id":"06427f1a8b158063","repo":"apache/iceberg","slug":"unrecognized-version-or-corrupt-state-06427f","errorCode":null,"errorMessage":"Unrecognized version or corrupt state: ","messagePattern":"Unrecognized version or corrupt state: ","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink/v2.3/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/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/MonitorSource.java#L184-L207","documentation":"TableChangeIterator's deserialize() checks a version byte written by serialize()/snapshot() into the Flink keyed state. If the stored byte array does not carry the current version marker, it assumes the state was produced by an incompatible (older/newer) job or is corrupt, and throws IOException so the state recovery fails fast instead of decoding garbage.","triggerScenarios":"Restoring a Flink job from a checkpoint/savepoint created by a different Iceberg/Flink version where the state format version changed; manually modifying or truncating state bytes; reading state serialized by code that wrote no version or a different one.","commonSituations":"Upgrading the Iceberg Flink runtime while resuming from an old savepoint; replaying a checkpoint across job migrations; state backend corruption after a crash.","solutions":["Take a savepoint with the same Iceberg version used at runtime, or restart the job from scratch (allowNotice/initial state) instead of restoring old state","Align the flink/iceberg dependency versions between the job that wrote the state and the one restoring it","Inspect the state backend entry to confirm the version byte; if corrupt, delete the offending state entry and reprocess","If upgrade is required, drain the maintenance stream (no in-flight state) before upgrading"],"exampleFix":"// before: restoring old savepoint with new connector\nflink run -s old-savepoint ...  // throws Unrecognized version or corrupt state: 1\n// after: use a savepoint/checkpoint taken with matching iceberg-flink-runtime, or start fresh\nflink run -s checkpoint-made-with-same-version ...","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  iterator = TableChangeIterator.deserialize(...);\n} catch (IOException e) {\n  LOG.error(\"State version incompatible; restart without restoring old state\", e);\n  throw e; // fail the restore; do not silently reprocess with wrong format\n}","preventionTips":["Restore checkpoints/savepoints only with the same iceberg-flink-runtime version that wrote them","Pin Iceberg version across the whole pipeline to avoid mixed state formats","Monitor upgrade paths: drain state before upgrading major versions"],"tags":["flink","serialization","checkpoint-restore","state-compatibility"],"backgroundTag":"invalid-config-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"}