{"record":{"id":"773a25accac80c25","repo":"apache/beam","slug":"source-must-be-an-unboundedsource-got-r","errorCode":null,"errorMessage":"source must be an UnboundedSource, got %r","messagePattern":"source must be an UnboundedSource, got %r","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/unbounded_source.py","lineNumber":945,"sourceCode":"    poll_interval: resume delay in seconds applied when the reader has no data,\n      which bounds how often an idle source is polled. Must be >= 0.\n    max_records_per_bundle: a busy reader self-checkpoints after emitting this\n      many records in one bundle. Must be >= 1. Defaults to 10000.\n    max_read_time_seconds: a busy reader self-checkpoints after this many\n      seconds in one bundle. Must be > 0. Defaults to 10.0. The deadline is\n      checked between records, so a reader that blocks inside ``advance()`` may\n      overrun it; ``max_records_per_bundle`` is the hard backstop.\n\n  The bundle self-checkpoints as soon as either cap is reached.\n  \"\"\"\n  def __init__(\n      self,\n      source: UnboundedSource,\n      poll_interval: float = _DEFAULT_POLL_INTERVAL_SECONDS,\n      max_records_per_bundle: int = _DEFAULT_MAX_RECORDS_PER_BUNDLE,\n      max_read_time_seconds: float = _DEFAULT_MAX_READ_TIME_SECONDS):\n    if not isinstance(source, UnboundedSource):\n      raise TypeError('source must be an UnboundedSource, got %r' % (source, ))\n    if max_records_per_bundle < 1:\n      raise ValueError(\n          'max_records_per_bundle must be >= 1, got %r' %\n          (max_records_per_bundle, ))\n    if max_read_time_seconds <= 0:\n      raise ValueError(\n          'max_read_time_seconds must be > 0, got %r' %\n          (max_read_time_seconds, ))\n    if poll_interval < 0:\n      raise ValueError(\n          'poll_interval must be >= 0, got %r' % (poll_interval, ))\n    super().__init__()\n    self._source = source\n    self._poll_interval = poll_interval\n    self._max_records_per_bundle = max_records_per_bundle\n    self._max_read_time_seconds = max_read_time_seconds\n\n  def expand(self, pbegin):","sourceCodeStart":927,"sourceCodeEnd":963,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/unbounded_source.py#L927-L963","documentation":"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.","triggerScenarios":"Constructing the reader with a BoundedSource, a string/URI, a class instead of an instance, or any non-UnboundedSource object passed as source.","commonSituations":"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.","solutions":["Pass an instance of a class that inherits from UnboundedSource (call the constructor instead of passing the class)","Verify no wrapper or options object is being passed where the source itself is expected","Check the argument order of the constructor so the source parameter is not receiving another positional argument"],"exampleFix":"# before\nreader = UnboundedSourceReader('kafka:9092')\n# after\nreader = UnboundedSourceReader(KafkaUnboundedSource(bootstrap='kafka:9092'))","handlingStrategy":"type-guard","validationCode":"if not isinstance(source, UnboundedSource):\n    raise TypeError('source must be an UnboundedSource, got %r' % type(source))","typeGuard":"def is_unbounded_source(x) -> bool:\n    return isinstance(x, UnboundedSource)","tryCatchPattern":"try:\n    reader = Reader(source)\nexcept TypeError:\n    source = resolve_unbounded_source(source)  # map config/wrapper to source instance\n    reader = Reader(source)","preventionTips":["Always instantiate the source class before passing it","Check constructor argument order when options are passed positionally","Annotate factory parameters as UnboundedSource so type checkers catch mistakes"],"tags":["python","apache-beam","constructor","type-error"],"backgroundTag":"invalid-constructor-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}