apache/beam · error · RuntimeError

Unable to claim position

Error message

Unable to claim position: %s

What it means

Raised by _TextSource-like TFRecord reading in apache_beam/io/tfrecordio.py when the OffsetRangeTracker refuses to claim the current offset during sequential record iteration. RangeTracker.try_claim returns False when the offset falls outside the restriction assigned to this bundle (e.g. the offset is below start or at/past stop), so the reader cannot proceed. Beam throws RuntimeError to abort the bundle rather than silently reading data owned by another bundle.

Solutions

  1. Verify the restriction passed to the OffsetRangeTracker matches the actual file offsets (no truncation/appending of the file between split and read).
  2. If you supplied a custom RangeTracker, ensure try_claim(claim) returns True for the current start position and any offset within [start, stop).
  3. Re-split or re-read the file with a full-range restriction (start=0, stop=OFFSET_INFINITY) to confirm the file itself reads cleanly.
  4. Check that checkpoint/resume offsets are taken from the same file generation (compare file size/mtime/ETag).

Example fix

// before
current_offset = offset_range_tracker.start_position()
while True:
  if not offset_range_tracker.try_claim(current_offset):
    raise RuntimeError('Unable to claim position: %s' % current_offset)

// after
# use a tracker whose restriction actually covers the offsets you will read
tracker = OffsetRangeTracker(0, file_size)  # not (resume_offset, resume_offset)
current_offset = tracker.start_position()
while True:
  if not tracker.try_claim(current_offset):
    break  # stop cleanly instead of raising when leaving the restriction
Defensive patterns

Strategy: try-catch

Validate before calling

start = offset_range_tracker.start_position()
stop = offset_range_tracker.stop_position()
assert start <= stop, 'invalid restriction: start %s > stop %s' % (start, stop)
# confirm the file size covers the restriction offsets
import os
assert offset_range_tracker.stop_position() == OFFSET_INFINITY or \
    offset_range_tracker.stop_position() <= os.path.getsize(file_name)

Type guard

def has_valid_restriction(tracker) -> bool:
    start = tracker.start_position()
    stop = getattr(tracker, 'stop_position', lambda: None)()
    return start is not None and (stop is None or start <= stop)

Try / catch

try:
  for record in source.read_records(file_name, tracker):
    process(record)
except RuntimeError as e:
  if 'Unable to claim position' in str(e):
    logger.error('range restriction mismatch for %s: %s', file_name, e)
    raise BeamIOError from e
  raise

Prevention

When it happens

Trigger: Calling read_records on a TFRecord file with an OffsetRangeTracker whose start_position is nonzero while the reader's internal invariant check ('Start position not 0') has passed but a subsequent try_claim(current_offset) fails — i.e. current_offset >= stop_position or < start_position of the tracker, typically from a corrupted or stale offset after resume, or a custom RangeTracker implementation with buggy claim semantics.

Common situations: Custom runners or resharding logic computing wrong split offsets for TFRecord sources; restoring from a checkpoint whose recorded offset exceeds the restriction stop; users implementing their own OffsetRangeTracker that returns False from try_claim for positions it already holds; file mutated/truncated between split and read so encoded record sizes shift offsets.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/c1ddb2ed42833772. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/io/tfrecordio.py:206

  def __init__(self, file_pattern, coder, compression_type, validate):
    """Initialize a TFRecordSource.  See ReadFromTFRecord for details."""
    super().__init__(
        file_pattern=file_pattern,
        compression_type=compression_type,
        splittable=False,
        validate=validate)
    self._coder = coder

  def read_records(self, file_name, offset_range_tracker):
    if offset_range_tracker.start_position():
      raise ValueError(
          'Start position not 0:%s' % offset_range_tracker.start_position())

    current_offset = offset_range_tracker.start_position()
    with self.open_file(file_name) as file_handle:
      while True:
        if not offset_range_tracker.try_claim(current_offset):
          raise RuntimeError('Unable to claim position: %s' % current_offset)
        record = _TFRecordUtil.read_record(file_handle)
        if record is None:
          return  # Reached EOF
        else:
          current_offset += _TFRecordUtil.encoded_num_bytes(record)
          yield self._coder.decode(record)


def _create_tfrecordio_source(
    file_pattern=None, coder=None, compression_type=None):
  # We intentionally disable validation for ReadAll pattern so that reading does
  # not fail for globs (elements) that are empty.
  return _TFRecordSource(file_pattern, coder, compression_type, validate=False)


class ReadAllFromTFRecord(PTransform):
  """A ``PTransform`` for reading a ``PCollection`` of TFRecord files."""
  def __init__(

View on GitHub (pinned to 12126d8942)