apache/beam · error · ValueError

Invalid TestStream event

Error message

Invalid TestStream event: %s.

What it means

The DirectRunner's TestStream evaluator received an event object that is not one of the recognized TestStream event types (ElementEvent, WatermarkEvent, ProcessingTimeEvent); the evaluator cannot advance clock or watermarks for it, so it rejects the event during process_element.

Solutions

  1. Build events only via TestStream's Builder API (add_elements, advance_watermark_to, advance_processing_time).
  2. Ensure the apache-beam SDK defining the event types matches the runner version.
  3. Check the event object's type; only ElementEvent/WatermarkEvent/ProcessingTimeEvent are accepted.
  4. If you authored a custom event type, extend _TestStreamEvaluator.process_element to handle it.

Example fix

# before
test_stream._events.append(MyCustomEvent())
# after
test_stream = (test_stream.TestStream()
    .advance_watermark_to(infinity)
    .add_elements(['a']))
Defensive patterns

Strategy: type-guard

Validate before calling

from apache_beam.testing.test_stream import ElementEvent, WatermarkEvent, ProcessingTimeEvent
assert isinstance(event, (ElementEvent, WatermarkEvent, ProcessingTimeEvent))

Type guard

def is_valid_teststream_event(event) -> bool:
    from apache_beam.testing.test_stream import ElementEvent, WatermarkEvent, ProcessingTimeEvent
    return isinstance(event, (ElementEvent, WatermarkEvent, ProcessingTimeEvent))

Try / catch

try:
    run_test()
except ValueError as e:
    if 'Invalid TestStream event' in str(e):
        rebuild_stream_with_builder_api()

Prevention

When it happens

Trigger: Calling process_element on a _TestStreamEvaluator with an element that is not one of the recognized TestStream event types (ElementEvent, WatermarkEvent, ProcessingTimeEvent) — e.g. a raw value or a hand-rolled event class added after this runner code.

Common situations: Manually constructing TestStream events instead of using TestStream.Builder helpers; SDK/runner version mismatch introducing a new event type; mistakenly feeding regular pipeline elements into a TestStream evaluator in tests.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/direct/transform_evaluator.py:561

      # the elements. We chose to emit in the _WatermarkController so that the
      # element is emitted at the correct watermark value.
      if isinstance(event, (ElementEvent, WatermarkEvent)):
        # The WATERMARK_CONTROL_TAG is used to hold the _TestStream's
        # watermark to -inf, then +inf-1, then +inf. This watermark progression
        # is ultimately used to set up the proper holds to allow the
        # _WatermarkControllers to control their own output watermarks.
        if event.tag == _TestStream.WATERMARK_CONTROL_TAG:
          self.watermark = event.new_watermark
        else:
          main_output = list(self._outputs)[0]
          bundle = self._evaluation_context.create_bundle(main_output)
          bundle.output(GlobalWindows.windowed_value(event))
          self.bundles.append(bundle)
      elif isinstance(event, ProcessingTimeEvent):
        self._evaluation_context._watermark_manager._clock.advance_time(
            event.advance_by)
      else:
        raise ValueError('Invalid TestStream event: %s.' % event)

  def finish_bundle(self):
    unprocessed_bundles = []

    # Continue to send its own state to itself via an unprocessed bundle. This
    # acts as a heartbeat, where each element will read the next event from the
    # event stream.
    if not self.is_done:
      unprocessed_bundle = self._evaluation_context.create_bundle(
          pvalue.PBegin(self._applied_ptransform.transform.pipeline))
      unprocessed_bundle.add(
          GlobalWindows.windowed_value(b'', timestamp=self.watermark))
      unprocessed_bundles.append(unprocessed_bundle)

    # Returning the watermark in the dict here is used as a watermark hold.
    return TransformResult(
        self, self.bundles, unprocessed_bundles, None, {None: self.watermark})

View on GitHub (pinned to 12126d8942)