{"record":{"id":"d264728c37fe2d49","repo":"apache/iceberg","slug":"invalid-starting-record-offset-d-for-file-d-from-d26472","errorCode":null,"errorMessage":"Invalid starting record offset %d for file %d from CombinedScanTask: %s","messagePattern":"Invalid starting record offset (.+?) for file (.+?) from CombinedScanTask: (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java","lineNumber":98,"sourceCode":"    // skip files\n    Preconditions.checkState(\n        startingFileOffset < combinedTask.files().size(),\n        \"Invalid starting file offset %s for combined scan task with %s files: %s\",\n        startingFileOffset,\n        combinedTask.files().size(),\n        combinedTask);\n    for (long i = 0L; i < startingFileOffset; ++i) {\n      tasks.next();\n      fileOffset += 1;\n    }\n\n    updateCurrentIterator();\n    // skip records within the file\n    for (long i = 0; i < startingRecordOffset; ++i) {\n      if (currentFileHasNext() && hasNext()) {\n        next();\n      } else {\n        throw new IllegalStateException(\n            String.format(\n                Locale.ROOT,\n                \"Invalid starting record offset %d for file %d from CombinedScanTask: %s\",\n                startingRecordOffset,\n                startingFileOffset,\n                combinedTask));\n      }\n    }\n  }\n\n  @Override\n  public boolean hasNext() {\n    updateCurrentIterator();\n    return currentIterator.hasNext();\n  }\n\n  @Override\n  public T next() {","sourceCodeStart":80,"sourceCodeEnd":116,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java#L80-L116","documentation":"DataIterator.seek throws this IllegalStateException when the requested starting record offset exceeds the number of records actually available while skipping through the current file of a CombinedScanTask — i.e. the iterator ran out of records before consuming startingRecordOffset items. It indicates the offset (typically produced by the split/offset computation or residual filtering) is inconsistent with the file's row count. The message includes the offset, file index, and the full CombinedScanTask for diagnosis.","triggerScenarios":"Calling seek() with a startingRecordOffset larger than the file's record count, e.g. when combining a residual filter (delete files / filter residual rows) with a stale offset, or an off-by-one from a checkpointed reader position on a changed file set.","commonSituations":"Restoring Flink source state from a checkpoint whose task splits no longer match the current table snapshots (files rewritten by compaction), or constructing DataIterator manually with a wrong fileOffset/recordOffset pair.","solutions":["Ensure the CombinedScanTask used at seek time is the exact one the offset was computed against (same snapshot).","Recompute the starting offset against the current snapshot instead of reusing stale checkpoint offsets after table maintenance (compaction/rewrite).","Verify position merging logic (file offset + record offset) when implementing custom source resume logic.","Catch IllegalStateException around DataIterator creation and fall back to re-reading the split from its start."],"exampleFix":"// before: stale offset after compaction\nlong offset = previousState.startingRecordOffset; // computed on old files\niterator.seek(fileOffset, offset);\n// after: validate offset against current file\nlong fileRecords = asFilePaths(currentFile).positionCount();\nif (offset >= fileRecords) {\n  offset = 0; // or recompute from current snapshot\n}\niterator.seek(fileOffset, offset);","handlingStrategy":"validation","validationCode":"// validate offset before seek\nlong available = /* records in the current file for this split */;\nif (startingRecordOffset >= available) {\n  throw new IllegalArgumentException(\"Offset \" + startingRecordOffset + \" exceeds file records \" + available);\n}","typeGuard":null,"tryCatchPattern":"try {\n  dataIterator.seek(fileOffset, recordOffset);\n} catch (IllegalStateException e) {\n  LOG.warn(\"Stale offset; re-reading split from start\", e);\n  dataIterator = newDataIterator(combinedTask);\n}","preventionTips":["Always pair offsets with the exact CombinedScanTask/snapshot they were computed from.","Recompute offsets after table maintenance operations like compaction or rewrites.","Store the snapshot ID alongside checkpointed read positions and validate on restore."],"tags":["flink","scan-planning","state-restore"],"backgroundTag":"value-out-of-range","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}