What causes "NotImplementedError" in apache/beam?

Override __call__ in your PollFn subclass and return a PollResult (e.g. PollResult([items], timestamps) or None to signal end)

apache/beam · error · NotImplementedError

NotImplementedError

Error message

NotImplementedError

What it means

PollFn is an abstract callable interface for the watch() source: __call__(element) must return a PollResult describing one poll round. The base class raises NotImplementedError to signal that subclasses must override __call__; it is a pure contract, not a runtime failure of the library.

Source

Thrown at sdks/python/apache_beam/io/watch.py:192

  """Optional base for a poll function ``input -> PollResult``.

  Any callable with that signature works; subclass only to attach an output
  coder hint via :meth:`default_output_coder`::

      from apache_beam import coders

      class ListFiles(PollFn):
        def __call__(self, prefix):
          return PollResult.incomplete(list_files(prefix))

        def default_output_coder(self):
          return coders.StrUtf8Coder()

  A plain function can instead annotate its return type as ``PollResult[V]``
  and have the output coder inferred from ``V``.
  """
  def __call__(self, element: Any) -> PollResult:
    raise NotImplementedError

  def default_output_coder(self) -> Optional[Coder]:
    return None


class TerminationCondition(object):
  """Per-input stop policy with immutable, encodable state.

  Hooks follow the lifecycle of one input's polling loop. ``state`` flows from
  :meth:`for_new_input` through the per-round hooks and is serialized with
  :meth:`state_coder`.
  """
  def for_new_input(self, now: Timestamp, element: Any) -> Any:
    raise NotImplementedError

  def on_seen_new_output(self, now: Timestamp, state: Any) -> Any:
    return state

View on GitHub (pinned to 12126d8942)

Solutions

Override __call__ in your PollFn subclass and return a PollResult (e.g. PollResult([items], timestamps) or None to signal end)
If you have a plain function, annotate its return type as PollResult[V] and pass the function instead of a PollFn subclass
Verify the object passed to watch() is the subclass instance, not the abstract base

Example fix

# before
class MyPoll(PollFn):
    def poll(self, element):
        return PollResult(fetch(element))
# after
class MyPoll(PollFn):
    def __call__(self, element):
        return PollResult(fetch(element))

Defensive patterns

Strategy: type-guard

Validate before calling

poll = poll_fn()
assert type(poll).__call__ is not PollFn.__call__, 'PollFn subclass must override __call__'

Type guard

def has_call_impl(fn) -> bool:
    return isinstance(fn, PollFn) and type(fn).__call__ is not PollFn.__call__

Try / catch

try:
    _ = poll_fn(element)
except NotImplementedError:
    raise TypeError('pass a PollFn subclass overriding __call__ or a typed function')

Prevention

Name the override __call__ exactly (not poll or run)
In quick tests, instantiate the concrete subclass, never the base class
Prefer plain functions annotated with PollResult[V] return types when possible

When it happens

Trigger: Defining a custom PollFn subclass without overriding __call__, or passing a bare PollFn/object lacking a __call__ implementation into watch() so the base implementation executes.

Common situations: Subclassing PollFn but naming the method wrongly (e.g. poll instead of __call__); instantiating the abstract base directly in tests; forgetting __call__ when implementing only default_output_coder.

Related errors

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