{"record":{"id":"fe79ad9293902cf0","repo":"apache/iceberg","slug":"unknown-version","errorCode":null,"errorMessage":"Unknown version: ","messagePattern":"Unknown version: ","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/IcebergEnumeratorPositionSerializer.java","lineNumber":53,"sourceCode":"      ThreadLocal.withInitial(() -> new DataOutputSerializer(128));\n\n  @Override\n  public int getVersion() {\n    return VERSION;\n  }\n\n  @Override\n  public byte[] serialize(IcebergEnumeratorPosition position) throws IOException {\n    return serializeV1(position);\n  }\n\n  @Override\n  public IcebergEnumeratorPosition deserialize(int version, byte[] serialized) throws IOException {\n    switch (version) {\n      case 1:\n        return deserializeV1(serialized);\n      default:\n        throw new IOException(\"Unknown version: \" + version);\n    }\n  }\n\n  private byte[] serializeV1(IcebergEnumeratorPosition position) throws IOException {\n    DataOutputSerializer out = SERIALIZER_CACHE.get();\n    out.writeBoolean(position.snapshotId() != null);\n    if (position.snapshotId() != null) {\n      out.writeLong(position.snapshotId());\n    }\n    out.writeBoolean(position.snapshotTimestampMs() != null);\n    if (position.snapshotTimestampMs() != null) {\n      out.writeLong(position.snapshotTimestampMs());\n    }\n    byte[] result = out.getCopyOfBuffer();\n    out.clear();\n    return result;\n  }\n","sourceCodeStart":35,"sourceCodeEnd":71,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/IcebergEnumeratorPositionSerializer.java#L35-L71","documentation":"IcebergEnumeratorPositionSerializer.deserialize only supports serialization version 1; encountering a different version byte in checkpoint state throws IOException. The serialized enumerator position was written by an incompatible serializer version.","triggerScenarios":"Restoring a Flink checkpoint/savepoint whose enumerator state bytes were written with a version other than 1 (e.g., newer connector restoring under an older connector).","commonSituations":"Downgrading the Iceberg Flink connector after upgrading; restoring state across incompatible connector releases.","solutions":["Restore the checkpoint/savepoint with the same or newer connector version that wrote it.","Upgrade the connector rather than downgrading, or start a fresh job without state.","If downgrade is required, let the job run from a new checkpoint written by the target version first."],"exampleFix":"// before\n# downgrade connector to v1.20 older patch and restore new checkpoint\n// after\n# keep connector at the version that wrote the checkpoint, or restart without state","handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  position = positionSerializer.deserialize(version, bytes);\n} catch (IOException e) {\n  LOG.warn(\"Incompatible enumerator position state, starting fresh\", e);\n  position = null; // fall back to configured starting strategy\n}","preventionTips":["Never downgrade the connector across checkpoints","Take savepoints before upgrades and test restore with the new version","Keep state serializer versions in sync within one connector release line"],"tags":["flink","serialization","checkpoint","version-mismatch"],"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"}