apache/beam · error · NotImplementedError

Should only be called on sub-trackers

Error message

Should only be called on sub-trackers

What it means

_ConcatSourceProgressTracker represents progress across multiple sub-sources; individual position tracking must be done by each sub-tracker. set_current_position() is intentionally unimplemented on the composite tracker and always raises NotImplementedError('Should only be called on sub-trackers').

Solutions

1. Route set_current_position calls to the appropriate sub-tracker for the current source index
2. Use position_at_fraction()/fraction-based APIs on the composite tracker instead
3. Update runner integration to handle composite trackers specially

Example fix

// before
reader.get_progress().set_current_position(pos)
// after
sub_tracker = tracker.trackers[current_source_index]
sub_tracker.set_current_position(local_pos)

from apache_beam.io.concat_source import _ConcatSourceProgressTracker
if isinstance(tracker, _ConcatSourceProgressTracker):
    raise TypeError('use sub-trackers for set_current_position')

def is_sub_tracker(tracker):
    return not hasattr(tracker, '_source_bundles')

try:
    tracker.set_current_position(pos)
except NotImplementedError:
    sub = tracker.get_sub_trackers()[current_index]
    sub.set_current_position(local_pos)

When it happens

Trigger: Calling set_current_position(pos) on the ConcatSource-level progress tracker (obtained from a ConcatSource reader) instead of on a sub-source tracker; custom runner code that drives progress generically.

Common situations: Custom runners/dynamic work rebalancing that assume every tracker supports set_current_position; unit tests exercising composite source progress directly.

Related errors

• A dataSourceConfiguration or dataSourceProviderFn has…
• A list of URNs for overriding transforms was provided but…
• A cannot be expanded
• A transform cannot be initiated using the provided config…
• Accumulators not supported in this context

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

Appendix: source

Thrown at sdks/python/apache_beam/io/concat_source.py:214

          # Prefer to split on even boundary.
          split_pos = None
          ratio = self._cumulative_weights[source_ix]
        else:
          # Split the current subsource.
          split = self.sub_range_tracker(source_ix).try_split(source_pos)
          if not split:
            return None
          split_pos, frac = split
          ratio = self.local_to_global(source_ix, frac)

        self._end = source_ix, split_pos
        self._cumulative_weights = [
            min(w / ratio, 1) for w in self._cumulative_weights
        ]
        return (source_ix, split_pos), ratio

  def set_current_position(self, pos):
    raise NotImplementedError('Should only be called on sub-trackers')

  def position_at_fraction(self, fraction):
    source_ix, source_frac = self.global_to_local(fraction)
    last = self._end[0] if self._end[1] is None else self._end[0] + 1
    if source_ix == last:
      return (source_ix, None)
    else:
      return (
          source_ix,
          self.sub_range_tracker(source_ix).position_at_fraction(source_frac))

  def fraction_consumed(self):
    with self._lock:
      if self._claimed_source_ix == len(self._source_bundles):
        return 1.0
      else:
        return self.local_to_global(
            self._claimed_source_ix,

View on GitHub (pinned to 12126d8942)