apache/beam · error · NotImplementedError

unimplemented

Error message

unimplemented

What it means

`add_state` on this trigger manager raises NotImplementedError('unimplemented') because trigger state can only be kept in per-window context in the Fn API runner. The base/non-windowed TriggerContext cannot store arbitrary stateful values keyed by tag. This is a deliberate limitation, not a bug: callers must use PerWindowTriggerContext instead.

Solutions

1. Replace the flow with one that operates in per-window context (PerWindowTriggerContext), where state is supported
2. Use a trigger that does not require non-window state, or downgrade/adjust the trigger spec
3. Track upstream: check apache_beam releases for per-window state support of your trigger
4. Implement add_state/get_state/clear_state in a custom TriggerContext subclass if you control the runner wiring

Example fix

// before
ctx.add_state(tag, value)
// after
per_window_ctx = PerWindowTriggerContext(window, fn_runner_ctx)
per_window_ctx.add_state(tag, value)

if type(ctx).__module__.endswith('trigger_manager') and not hasattr(ctx, 'window'):
  raise ValueError('add_state requires a per-window TriggerContext')

def is_per_window_ctx(ctx):
  return hasattr(ctx, 'window') and hasattr(ctx, 'parent')

try:
  ctx.add_state(tag, value)
except NotImplementedError:
  # fall back to per-window context or skip stateful trigger logic
  per_window_ctx.add_state(tag, value)

When it happens

Trigger: Calling `trigger_manager.add_state(tag, value)` (via a TriggerContext that is not window-scoped) while a trigger driver tries to persist state for a trigger that requires non-window state.

Common situations: Custom or newly used Beam triggers (e.g. watermark-based or repeat triggers) that request state storage outside a window; running pipelines on the Python Fn API runner whose trigger spec requires generic state.

Related errors

• accumulation_mode must be provided for non-trivial triggers
• Assigning an index is not yet supported. Consider using…
• by
• collecting metrics will come later!
• concat(ignore_index)

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/fn_api_runner/trigger_manager.py:398

                          for (window, state_tag, state) in all_triplets
                          if window not in to_be_merged] + merging_away_triplets
    self.window_tag_values.clear()
    for t in resulting_triplets:
      self.window_tag_values.add(t)

    # Merge also element-window pairs
    all_elements = self.all_elements_state.read()
    resulting_elements = [
        (merge_result if e[0] in to_be_merged else e[0], e[1])
        for e in all_elements
    ]
    self.all_elements_state.clear()
    for e in resulting_elements:
      self.all_elements_state.add(e)

  def add_state(self, tag, value):
    # State can only be kept in per-window context, so this is not implemented.
    raise NotImplementedError('unimplemented')

  def get_state(self, tag):
    # State can only be kept in per-window context, so this is not implemented.
    raise NotImplementedError('unimplemented')

  def clear_state(self, tag):
    # State can only be kept in per-window context, so this is not implemented.
    raise NotImplementedError('unimplemented')


class PerWindowTriggerContext(TriggerContext):
  def __init__(self, window, parent: FnRunnerStatefulTriggerContext):
    self.window = window
    self.parent = parent

  def get_current_time(self):
    return self.parent.get_current_time()

View on GitHub (pinned to 12126d8942)