apache/beam · error · TypeError

source must be an UnboundedSource, got %r

Error message

source must be an UnboundedSource, got %r

What it means

The constructor for this unbounded-source read provider requires the source argument to be an UnboundedSource instance. It raises TypeError immediately so an incorrectly typed source fails at construction time rather than later during pipeline expansion. This guards the polling/bundle-reading machinery which depends on UnboundedSource-specific methods.

Source

Thrown at sdks/python/apache_beam/io/unbounded_source.py:945

    poll_interval: resume delay in seconds applied when the reader has no data,
      which bounds how often an idle source is polled. Must be >= 0.
    max_records_per_bundle: a busy reader self-checkpoints after emitting this
      many records in one bundle. Must be >= 1. Defaults to 10000.
    max_read_time_seconds: a busy reader self-checkpoints after this many
      seconds in one bundle. Must be > 0. Defaults to 10.0. The deadline is
      checked between records, so a reader that blocks inside ``advance()`` may
      overrun it; ``max_records_per_bundle`` is the hard backstop.

  The bundle self-checkpoints as soon as either cap is reached.
  """
  def __init__(
      self,
      source: UnboundedSource,
      poll_interval: float = _DEFAULT_POLL_INTERVAL_SECONDS,
      max_records_per_bundle: int = _DEFAULT_MAX_RECORDS_PER_BUNDLE,
      max_read_time_seconds: float = _DEFAULT_MAX_READ_TIME_SECONDS):
    if not isinstance(source, UnboundedSource):
      raise TypeError('source must be an UnboundedSource, got %r' % (source, ))
    if max_records_per_bundle < 1:
      raise ValueError(
          'max_records_per_bundle must be >= 1, got %r' %
          (max_records_per_bundle, ))
    if max_read_time_seconds <= 0:
      raise ValueError(
          'max_read_time_seconds must be > 0, got %r' %
          (max_read_time_seconds, ))
    if poll_interval < 0:
      raise ValueError(
          'poll_interval must be >= 0, got %r' % (poll_interval, ))
    super().__init__()
    self._source = source
    self._poll_interval = poll_interval
    self._max_records_per_bundle = max_records_per_bundle
    self._max_read_time_seconds = max_read_time_seconds

  def expand(self, pbegin):

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass an instance of a class that inherits from UnboundedSource (call the constructor instead of passing the class)
2. Verify no wrapper or options object is being passed where the source itself is expected
3. Check the argument order of the constructor so the source parameter is not receiving another positional argument

Example fix

# before
reader = UnboundedSourceReader('kafka:9092')
# after
reader = UnboundedSourceReader(KafkaUnboundedSource(bootstrap='kafka:9092'))

if not isinstance(source, UnboundedSource):
    raise TypeError('source must be an UnboundedSource, got %r' % type(source))

def is_unbounded_source(x) -> bool:
    return isinstance(x, UnboundedSource)

try:
    reader = Reader(source)
except TypeError:
    source = resolve_unbounded_source(source)  # map config/wrapper to source instance
    reader = Reader(source)

When it happens

Trigger: Constructing the reader with a BoundedSource, a string/URI, a class instead of an instance, or any non-UnboundedSource object passed as source.

Common situations: Passing Kafka/other connector options dicts instead of a configured source object; forgetting to instantiate the source class; migrating from bounded-source APIs that accept other element types.

Related errors

• element_size_fn must be callable
• Unable to convert objects of type %s to a PCollection
• Encountered unknown type {other!r}
• Proxy '{proxy}' has unsupported type '{type(proxy)}'
• Input should be a module object, got {str(module)} instead

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