{"record":{"id":"86f37119e34d524a","repo":"apache/beam","slug":"notimplementederror-unbounded-source","errorCode":null,"errorMessage":"NotImplementedError","messagePattern":"NotImplementedError","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/unbounded_source.py","lineNumber":180,"sourceCode":"    raised here is logged. On bundle retry an uncommitted mark may be re-cut\n    over an overlapping span, so this method must be idempotent (acknowledge by\n    absolute position).\n    \"\"\"\n    pass\n\n\nclass UnboundedReader(object):\n  \"\"\"Reads records from an :class:`UnboundedSource`.\n\n  Lifecycle: exactly one :meth:`start`, then any number of :meth:`advance`\n  calls; whenever one returns ``True`` the current record is available via\n  :meth:`get_current` / :meth:`get_current_timestamp`. A ``False`` return means\n  \"no data available right now\", which is distinct from end-of-stream: a reader\n  signals a permanent end by reporting a watermark of ``MAX_TIMESTAMP``.\n  \"\"\"\n  def start(self) -> bool:\n    \"\"\"Positions at the first record; returns whether one is available.\"\"\"\n    raise NotImplementedError\n\n  def advance(self) -> bool:\n    \"\"\"Advances to the next record. ``False`` means no data is available now.\n\n    Should not block. The wrapper enforces the per-bundle record and time caps\n    only between records, so a blocking ``start``/``advance`` can overrun the\n    time cap and stall the bundle. Return ``False`` when no data is currently\n    available instead of waiting.\n    \"\"\"\n    raise NotImplementedError\n\n  def get_current(self) -> Any:\n    \"\"\"Returns the record claimed by the last successful start/advance.\"\"\"\n    raise NotImplementedError\n\n  def get_current_timestamp(self) -> Timestamp:\n    \"\"\"Returns the event-time timestamp of the current record.\"\"\"\n    raise NotImplementedError","sourceCodeStart":162,"sourceCodeEnd":198,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/unbounded_source.py#L162-L198","documentation":"UnboundedReader.start() in apache_beam/io/unbounded_source.py is an abstract interface method that must be implemented by any subclass providing a streaming reader. It positions the reader at the first record and returns whether one is available. Python raises NotImplementedError because the concrete reader class inherits the stub instead of overriding it.","triggerScenarios":"Subclassing UnboundedReader (or a wrapper instantiating it) without implementing start(); the Beam pipeline calls start() when beginning a bundle and hits the stub body.","commonSituations":"Users building custom streaming sources for runners with the new unbounded-source API; partially migrated implementations where only advance()/get_watermark() were overridden; typos in the override name (e.g. 'starts') leaving the stub in place.","solutions":["Implement start() in your UnboundedReader subclass to position at the first record and return True/False.","Verify the method name and signature exactly match `def start(self) -> bool` (watch for typos or wrong casing).","If you only need a bounded source, use a BoundedSource instead of the unbounded interface so the stub is never invoked."],"exampleFix":"// before\nclass MyReader(UnboundedReader):\n  def advance(self) -> bool:\n    ...\n\n// after\nclass MyReader(UnboundedReader):\n  def start(self) -> bool:\n    self._it = iter(self._source.records)\n    return self.advance()\n  def advance(self) -> bool:\n    ...\n","handlingStrategy":"validation","validationCode":"from apache_beam.io.unbounded_source import UnboundedReader\nassert not getattr(reader.start, '__isabstractmethod__', False), 'UnboundedReader.start not implemented'","typeGuard":"import inspect\ndef reader_fully_implemented(cls) -> bool:\n    stubs = ['start', 'advance', 'get_current', 'get_current_timestamp', 'get_watermark', 'get_checkpoint_mark']\n    return all(not getattr(getattr(cls, m, None), '__isabstractmethod__', True) for m in stubs)","tryCatchPattern":"try:\n  available = reader.start()\nexcept NotImplementedError:\n  logger.error('%s does not implement UnboundedReader.start', type(reader).__name__)\n  raise","preventionTips":["Use an abstract base class (ABC) with @abstractmethod for custom readers so instantiation fails early.","Add a unit test asserting reader_fully_implemented(MyReader).","After Beam upgrades, diff the UnboundedReader interface against your subclass.","Check override names for typos with an editor's override/intellisense hints."],"tags":["apache-beam","python","abstract-method","streaming-source"],"backgroundTag":"abstract-method-not-implemented","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}