{"record":{"id":"027ee1188d66444b","repo":"apache/beam","slug":"invalid-teststream-event-s","errorCode":null,"errorMessage":"Invalid TestStream event: %s.","messagePattern":"Invalid TestStream event: (.+?)\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/direct/transform_evaluator.py","lineNumber":561,"sourceCode":"      # the elements. We chose to emit in the _WatermarkController so that the\n      # element is emitted at the correct watermark value.\n      if isinstance(event, (ElementEvent, WatermarkEvent)):\n        # The WATERMARK_CONTROL_TAG is used to hold the _TestStream's\n        # watermark to -inf, then +inf-1, then +inf. This watermark progression\n        # is ultimately used to set up the proper holds to allow the\n        # _WatermarkControllers to control their own output watermarks.\n        if event.tag == _TestStream.WATERMARK_CONTROL_TAG:\n          self.watermark = event.new_watermark\n        else:\n          main_output = list(self._outputs)[0]\n          bundle = self._evaluation_context.create_bundle(main_output)\n          bundle.output(GlobalWindows.windowed_value(event))\n          self.bundles.append(bundle)\n      elif isinstance(event, ProcessingTimeEvent):\n        self._evaluation_context._watermark_manager._clock.advance_time(\n            event.advance_by)\n      else:\n        raise ValueError('Invalid TestStream event: %s.' % event)\n\n  def finish_bundle(self):\n    unprocessed_bundles = []\n\n    # Continue to send its own state to itself via an unprocessed bundle. This\n    # acts as a heartbeat, where each element will read the next event from the\n    # event stream.\n    if not self.is_done:\n      unprocessed_bundle = self._evaluation_context.create_bundle(\n          pvalue.PBegin(self._applied_ptransform.transform.pipeline))\n      unprocessed_bundle.add(\n          GlobalWindows.windowed_value(b'', timestamp=self.watermark))\n      unprocessed_bundles.append(unprocessed_bundle)\n\n    # Returning the watermark in the dict here is used as a watermark hold.\n    return TransformResult(\n        self, self.bundles, unprocessed_bundles, None, {None: self.watermark})\n","sourceCodeStart":543,"sourceCodeEnd":579,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/direct/transform_evaluator.py#L543-L579","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Build events only via TestStream's Builder API (add_elements, advance_watermark_to, advance_processing_time).","Ensure the apache-beam SDK defining the event types matches the runner version.","Check the event object's type; only ElementEvent/WatermarkEvent/ProcessingTimeEvent are accepted.","If you authored a custom event type, extend _TestStreamEvaluator.process_element to handle it."],"exampleFix":"# before\ntest_stream._events.append(MyCustomEvent())\n# after\ntest_stream = (test_stream.TestStream()\n    .advance_watermark_to(infinity)\n    .add_elements(['a']))","handlingStrategy":"type-guard","validationCode":"from apache_beam.testing.test_stream import ElementEvent, WatermarkEvent, ProcessingTimeEvent\nassert isinstance(event, (ElementEvent, WatermarkEvent, ProcessingTimeEvent))","typeGuard":"def is_valid_teststream_event(event) -> bool:\n    from apache_beam.testing.test_stream import ElementEvent, WatermarkEvent, ProcessingTimeEvent\n    return isinstance(event, (ElementEvent, WatermarkEvent, ProcessingTimeEvent))","tryCatchPattern":"try:\n    run_test()\nexcept ValueError as e:\n    if 'Invalid TestStream event' in str(e):\n        rebuild_stream_with_builder_api()","preventionTips":["Only construct TestStream events via TestStream's Builder methods.","Keep SDK and test dependencies version-aligned.","Never feed raw pipeline elements into TestStream evaluators."],"tags":["python","apache-beam","teststream","testing"],"backgroundTag":"invalid-enum-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}