apache/beam · error · NotImplementedError

NotImplementedError

Error message

NotImplementedError

What it means

BaseTimer.clear is an abstract-style placeholder in apache_beam.transforms.userstate that unconditionally raises NotImplementedError. It defines the timer interface; only concrete runtime implementations (e.g. RuntimeTimer in the direct runner) provide a body. Calling clear() on BaseTimer directly means you are using the interface class instead of a runtime-backed timer.

Solutions

1. Use the runner-provided runtime timer (RuntimeTimer) rather than BaseTimer directly.
2. In tests, inject a fake subclass that overrides clear() instead of instantiating BaseTimer.
3. If writing a runner/integration, implement clear() in your BaseTimer subclass.

Example fix

# before
timer = BaseTimer(); timer.clear()
# after
class FakeTimer(BaseTimer):
  def clear(self, dynamic_timer_tag='') -> None: self.cleared = True
timer = FakeTimer(); timer.clear()

assert isinstance(timer, BaseTimer) and type(timer) is not BaseTimer, 'use a runtime timer implementation'

def is_usable_timer(t) -> bool:
  return isinstance(t, BaseTimer) and type(t).clear is not BaseTimer.clear

try:
  timer.clear()
except NotImplementedError as e:
  logging.error('timer backend does not implement clear: %s', e)

When it happens

Trigger: Calling clear() on a BaseTimer instance obtained outside a running pipeline context (e.g. instantiating BaseTimer manually in tests or custom code) instead of using the runner-provided RuntimeTimer.

Common situations: Unit-testing a DoFn's timer methods by hand-constructing timer objects; custom runner integrations that forgot to subclass BaseTimer and override clear; calling clear before the DoFn's timer context is set up.

Related errors

• Assigning an index is not yet supported. Consider using…
• by
• collecting metrics will come later!
• concat(ignore_index)
• concat(levels)

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/userstate.py:327

  # Reject DoFns that use timer specs without corresponding timer callbacks.
  for timer_spec in all_timer_specs:
    if not timer_spec._attached_callback:
      raise ValueError((
          'DoFn %r has a TimerSpec without an associated on_timer '
          'callback: %s.') % (dofn, timer_spec))
    method_name = timer_spec._attached_callback.__name__
    if (timer_spec._attached_callback != getattr(dofn, method_name,
                                                 None).__func__):  # type: ignore[union-attr]
      raise ValueError((
          'The on_timer callback for %s is not the specified .%s method '
          'for DoFn %r (perhaps it was overwritten?).') %
                       (timer_spec, method_name, dofn))


class BaseTimer(object):
  def clear(self, dynamic_timer_tag: str = '') -> None:
    raise NotImplementedError

  def set(self, timestamp: Timestamp, dynamic_timer_tag: str = '') -> None:
    raise NotImplementedError


_TimerTuple = collections.namedtuple('timer_tuple', ('cleared', 'timestamp'))  # type: ignore[name-match]


class RuntimeTimer(BaseTimer):
  """Timer interface object passed to user code."""
  def __init__(self) -> None:
    self._timer_recordings: dict[str, _TimerTuple] = {}
    self._cleared = False
    self._new_timestamp: Optional[Timestamp] = None

  def clear(self, dynamic_timer_tag: str = '') -> None:
    self._timer_recordings[dynamic_timer_tag] = _TimerTuple(
        cleared=True, timestamp=None)

View on GitHub (pinned to 12126d8942)