apache/beam · error · ValueError

accumulation_mode must be provided for non-trivial triggers

Error message

accumulation_mode must be provided for non-trivial triggers

What it means

Raised in the WindowInto/Windowing constructor when a custom (non-default) trigger is supplied but accumulation_mode is not specified. Beam must know whether fired panes should accumulate or discard contents when a trigger fires multiple times; this is only inferable for the DefaultTrigger (which defaults to DISCARDING).

Solutions

  1. Add accumulation_mode=beam.trigger.AccumulationMode.DISCARDING (or ACCUMULATING) to the WindowInto call.
  2. Use ACCUMULATING if later panes should contain all data since window start; DISCARDING if only new data.
  3. If no custom trigger is actually needed, remove the trigger argument so the default applies.

Example fix

// before
pc | beam.WindowInto(beam.trigger.AfterCount(5))
// after
pc | beam.WindowInto(beam.trigger.AfterCount(5), accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING)
Defensive patterns

Strategy: validation

Validate before calling

trigger = beam.trigger.AfterCount(5)
assert accumulation_mode is not None or isinstance(trigger, beam.trigger.DefaultTrigger), 'set accumulation_mode with custom triggers'

Try / catch

try:
    pc | beam.WindowInto(trigger, accumulation_mode=mode)
except ValueError as e:
    log.error('Windowing config error: %s', e)

Prevention

When it happens

Trigger: beam.WindowInto(beam.trigger.AfterCount(10)) or any custom trigger (AfterWatermark, AfterProcessingTime, Repeatedly, etc.) without accumulation_mode=..., e.g. beam.WindowInto(AfterWatermark(), windowfn=...).

Common situations: Upgrading Beam where triggers were used without accumulation mode; copying examples of custom triggers that omit the parameter; adding a trigger to existing WindowInto code that previously used defaults.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/core.py:3891

      allowed_lateness: Maximum delay in seconds after end of window
        allowed for any late data to be processed without being discarded
        directly.
      environment_id: Environment where the current window_fn should be
        applied in.
    """
    global AccumulationMode, DefaultTrigger  # pylint: disable=global-variable-not-assigned
    # pylint: disable=wrong-import-order, wrong-import-position
    from apache_beam.transforms.trigger import AccumulationMode
    from apache_beam.transforms.trigger import DefaultTrigger

    # pylint: enable=wrong-import-order, wrong-import-position
    if triggerfn is None:
      triggerfn = DefaultTrigger()
    if accumulation_mode is None:
      if triggerfn == DefaultTrigger():
        accumulation_mode = AccumulationMode.DISCARDING
      else:
        raise ValueError(
            'accumulation_mode must be provided for non-trivial triggers')
    if not windowfn.get_window_coder().is_deterministic():
      raise ValueError(
          'window fn (%s) does not have a determanistic coder (%s)' %
          (windowfn, windowfn.get_window_coder()))
    self.windowfn = windowfn
    self.triggerfn = triggerfn
    self.accumulation_mode = accumulation_mode
    self.allowed_lateness = Duration.of(allowed_lateness)
    self.environment_id = environment_id
    self.timestamp_combiner = (
        timestamp_combiner or TimestampCombiner.OUTPUT_AT_EOW)
    self._is_default = (
        self.windowfn == GlobalWindows() and
        self.triggerfn == DefaultTrigger() and
        self.accumulation_mode == AccumulationMode.DISCARDING and
        self.timestamp_combiner == TimestampCombiner.OUTPUT_AT_EOW and
        self.allowed_lateness == 0)

View on GitHub (pinned to 12126d8942)