{"record":{"id":"b35ed995619f8684","repo":"apache/iceberg","slug":"unrecognized-version-or-corrupt-state","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.1/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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/MonitorSource.java#L184-L207","documentation":"TableChangeIterator's serializer in MonitorSource stores a single long (last snapshot id) tagged with a version byte. On deserialize, if the stored state's version is not CURRENT_VERSION, the state is neither readable nor migratable, so an IOException 'Unrecognized version or corrupt state' is thrown.","triggerScenarios":"Restoring a Flink source whose serializer state was written by an older Iceberg version (different CURRENT_VERSION) or whose state bytes are truncated/corrupt in the checkpoint/savepoint.","commonSituations":"Upgrading Iceberg/Flink across a state-format version change while resuming from an old savepoint; corrupted checkpoint storage (e.g. object store truncation); manually edited or migrated operator state.","solutions":["Resume from a checkpoint/savepoint produced by the same (or compatible) Iceberg version that defines the same CURRENT_VERSION.","If the version change is intentional, allow the job to start fresh (drop the source state) — the monitor source will rescan from the current snapshot, possibly re-processing recent changes.","Verify checkpoint bytes are intact (no truncation in the state backend storage) before restoring."],"exampleFix":"// before\n// restoring savepoint from Iceberg 1.4 state into a build with CURRENT_VERSION=2 -> IOException\n// after\n// start the job without the old savepoint, or upgrade to the version that wrote it\nflink run -s <compatible-savepoint> ...","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  iterator = serializer.deserialize(version, bytes);\n} catch (IOException e) {\n  LOG.warn(\"Incompatible monitor-source state ({}); starting from current snapshot\", e.getMessage());\n  iterator = new TableChangeIterator(tableLoader, null, maxReadBack);\n}","preventionTips":["Resume only from checkpoints/savepoints written by a compatible Iceberg version","After upgrading Iceberg across a state format change, start the monitor source fresh","Verify checkpoint storage integrity (no truncation) before restore"],"tags":["flink","state-serialization","checkpoint","version-compatibility"],"backgroundTag":"json-unmarshal-failed","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"}