{"record":{"id":"59ebebb9eab27f55","repo":"apache/iceberg","slug":"unknown-version-59ebeb","errorCode":null,"errorMessage":"Unknown version: ","messagePattern":"Unknown version: ","errorType":"exception","errorClass":"IOException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/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/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/IcebergEnumeratorPositionSerializer.java#L35-L71","documentation":"IcebergEnumeratorPositionSerializer.deserialize switches on the serialization version; only version 1 is supported. If a checkpoint/serialized byte array carries a different version, it throws IOException('Unknown version: ' + version). This protects against restoring enumerator state written by an incompatible Iceberg version.","triggerScenarios":"Restoring a Flink job from a checkpoint/savepoint whose enumerator state was serialized with a different IcebergEnumeratorPositionSerializer version than the one in the current job jar.","commonSituations":"Upgrading or downgrading the iceberg-flink version and resuming from an old savepoint; mixing Iceberg jars across job upgrades; hand-editing or replaying serialized enumerator state.","solutions":["Resume from a checkpoint/savepoint created with the same (or compatible) Iceberg version as the running jar.","Upgrade iceberg-flink so the serializer recognizes the stored version, or downgrade to the previously used version.","If the state is unrecoverable, start the streaming job without state (e.g., with StartingStrategy.TABLE_SCAN_THEN_INCREMENTS) to re-plan from the latest snapshot.","Check the version byte of stored state to confirm which serializer version produced it before restoring."],"exampleFix":"// before\n// restore old savepoint with newer jar: serializer v2 state vs v1 reader -> Unknown version: 2\n// after\n// align versions, or restart stateless:\nScanContext ctx = ScanContext.builder().streamingStartingStrategy(StreamingStartingStrategy.TABLE_SCAN_THEN_INCREMENTS).build();","handlingStrategy":"fallback","validationCode":"// inspect checkpoint metadata for the serializer version before restoring\nint v = readEnumeratorStateVersion(checkpointHandle);\nif (v != 1) { LOG.warn(\"State written by serializer version {} is incompatible with this jar\", v); }","typeGuard":null,"tryCatchPattern":"try {\n  position = serializer.deserialize(version, bytes);\n} catch (IOException e) {\n  if (e.getMessage().startsWith(\"Unknown version\")) {\n    // fall back to stateless re-plan from latest snapshot\n    position = null;\n  } else { throw e; }\n}","preventionTips":["Restore only from checkpoints taken with the same iceberg-flink version.","When upgrading, prefer stateless restart of streaming sources or verify serializer version compatibility.","Keep a savepoint taken just before upgrades as a rollback path."],"tags":["flink","serialization","checkpoint-restore","version-skew"],"backgroundTag":"unsupported-version","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"}