apache/beam · error · ValueError

Trying to return a record

Error message

Trying to return a record [starting at %d] which is not greaterthan the last-attempted record [starting at %d]

What it means

OffsetRangeTracker.try_claim() enforces that record start positions are strictly monotonic. It raises ValueError when a caller (typically a Beam source's read loop) attempts to claim a record whose start offset is less than or equal to the last attempted record start, which would break progress tracking and resume semantics.

Solutions

1. Ensure the read loop monotonically advances the offset: only call try_claim with a record_start strictly greater than the previous one.
2. When resuming, initialize the tracker's start position to the last consumed offset (e.g. from range_tracker.start_position()) so re-reads are not claimed.
3. Check the offset computation (record size / cursor arithmetic) for an off-by-one that re-emits the same start.
4. If duplicates are expected from the source, filter/skip records whose start <= last attempted start before claiming.

Example fix

// before
for rec in records:
  if not range_tracker.try_claim(rec.start):
    break
  yield rec
// after
last_start = range_tracker.start_position()
for rec in records:
  if rec.start <= last_start:
    continue  # skip already-attempted records
  if not range_tracker.try_claim(rec.start):
    break
  last_start = rec.start
  yield rec

if record_start <= range_tracker._last_attempted_record_start:
  raise ValueError('record_start must be strictly greater than last attempted')

try:
  range_tracker.try_claim(record_start)
except ValueError as e:
  logging.warning('skipping non-monotonic claim: %s', e)

When it happens

Trigger: Calling try_claim(record_start) with record_start <= self._last_attempted_record_start, e.g. a custom bounded source re-yielding the same offset, an off-by-one loop that does not advance the iterator, or resuming reads from a stale checkpoint offset.

Common situations: Writing a custom iobase.BoundedSource/restriction provider with a buggy loop; re-reading records after a retry without advancing; source splitting code computing overlapping offsets between sub-ranges.

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

• A schema is required to write non-schema'd data.
• A sink must inherit iobase.Sink, iobase.NativeSink, or be a…
• An explicit schema is required to write non-schema'd…
• Cannot call read after iterating.
• Cannot create a temporary directory for root path prefix

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

Appendix: source

Thrown at sdks/python/apache_beam/io/range_trackers.py:121

          'last-returned record [starting at %d]' %
          (record_start, self._last_record_start))

    if (split_point and self._offset_of_last_split_point != -1 and
        record_start == self._offset_of_last_split_point):
      raise ValueError(
          'Record at a split point has same offset as the previous split '
          'point: %d' % record_start)

    if not split_point and self._last_record_start == -1:
      raise ValueError(
          'The first record [starting at %d] must be at a split point' %
          record_start)

  def try_claim(self, record_start):
    with self._lock:
      # Attempted claim should be monotonous.
      if record_start <= self._last_attempted_record_start:
        raise ValueError(
            'Trying to return a record [starting at %d] which is not greater'
            'than the last-attempted record [starting at %d]' %
            (record_start, self._last_attempted_record_start))
      self._validate_record_start(record_start, True)
      self._last_attempted_record_start = record_start
      if record_start >= self.stop_position():
        return False
      self._offset_of_last_split_point = record_start
      self._last_record_start = record_start
      self._split_points_seen += 1
      return True

  def set_current_position(self, record_start):
    with self._lock:
      self._validate_record_start(record_start, False)
      self._last_record_start = record_start

  def try_split(self, split_offset):

View on GitHub (pinned to 12126d8942)