apache/beam · error · RuntimeError

Finish Bundle should only output WindowedValue type but got

Error message

Finish Bundle should only output WindowedValue type but got %s

What it means

finish_bundle must emit only WindowedValue objects because bundle-finish outputs carry window/timestamp metadata needed downstream. _any_ other result type (including a raw TimestampedValue, str, dict, etc.) triggers this RuntimeError in finish_bundle_outputs.

Source

Thrown at sdks/python/apache_beam/runners/common.py:1879

    A value wrapped in a TaggedOutput object will be unwrapped and
    then dispatched to the appropriate indexed output.
    """
    if results is None:
      return

    for result in results:
      tag = None
      if isinstance(result, TaggedOutput):
        tag = result.tag
        if not isinstance(tag, str):
          raise TypeError('In %s, tag %s is not a string' % (self, tag))
        result = result.value

      if isinstance(result, WindowedValue):
        windowed_value = result
      else:
        raise RuntimeError('Finish Bundle should only output WindowedValue ' +\
                           'type but got %s' % type(result))

      if tag is None:
        self.main_receivers.receive(windowed_value)
      else:
        self.tagged_receivers[tag].receive(windowed_value)


class _NoContext(WindowFn.AssignContext):
  """An uninspectable WindowFn.AssignContext."""
  NO_VALUE = object()

  def __init__(self, value, timestamp=NO_VALUE):
    self.value = value
    self._timestamp = timestamp

  @property
  def timestamp(self):

View on GitHub (pinned to 12126d8942)

Solutions

Wrap each finish_bundle result: `yield WindowedValue(value, timestamp, [window])`, commonly via `beam.window.TimestampedValue(value, ts)` only in process — in finish_bundle construct WindowedValue directly.
Ensure timestamps are within allowed lateness/allowed timestamp bounds (windowing strategy).
Check the existing_windows of inputs (via DoFn.process context) and reuse them for outputs.

Example fix

// before
def finish_bundle(self):
    yield ('total', self.total)
// after
def finish_bundle(self):
    yield WindowedValue(('total', self.total), self.last_ts, self.window)

Defensive patterns

Strategy: validation

Validate before calling

assert all(isinstance(r, (WindowedValue, TaggedOutput)) for r in finish_bundle_results), 'finish_bundle must yield WindowedValue'

Type guard

def is_windowed_output(r):
    v = r.value if isinstance(r, TaggedOutput) else r
    return isinstance(v, WindowedValue)

Try / catch

try:
    invoker.invoke_finish_bundle()
except RuntimeError as e:
    if 'Finish Bundle should only output WindowedValue' in str(e):
        raise ValueError('Wrap finish_bundle outputs in WindowedValue') from e
    raise

Prevention

Always construct WindowedValue(value, timestamp, [window]) in finish_bundle.
Capture the window/timestamp from the last processed element to reuse in finish_bundle outputs.
Test finish_bundle paths with the DirectRunner where output validation runs.

When it happens

Trigger: finish_bundle yields plain values like `yield ('key', count)` or `yield TimestampedValue(v, ts)` instead of `yield WindowedValue(v, ts, window)`.

Common situations: Writing aggregation/summary logic in finish_bundle and forgetting to wrap results with beam.window.TimestampedValue(...).in_window(...) / WindowedValue; assuming Beam auto-wraps finish_bundle outputs like it does for process().

Related errors

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