apache/beam · error · NotImplementedError

{transform.spec.urn}

Error message

{transform.spec.urn}

What it means

Within check_requirements(), if a transform has the TEST_STREAM primitive URN but the runner's supported_requirements does not include TestStream, a NotImplementedError with the URN itself is raised. TestStream requires explicit runner support for deterministic testing of watermark/timer behavior.

Solutions

  1. Run TestStream pipelines with a runner that supports TestStream (e.g. the FnApiRunner/test runner)
  2. Add TEST_STREAM to the runner's supported_requirements if you own the runner and implement support
  3. Replace TestStream with a regular test harness when runner support is unavailable

Example fix

// before
with beam.Pipeline(runner='FlinkRunner') as p:
  _ = (p | TestStream().add_elements([...]))
// after
with beam.Pipeline(runner='FnApiRunner') as p:
  _ = (p | TestStream().add_elements([...]))
Defensive patterns

Strategy: try-catch

Validate before calling

from apache_beam.coders import coder_impl
from apache_beam.portability.api import beam_runner_api_pb2
uses_test_stream = any(t.spec.urn == 'beam:transform:test_stream:v1'
                       for t in pipeline_proto.components.transforms.values())

Try / catch

try:
    runner.check_requirements(pipeline_proto, runner.supported_requirements)
except NotImplementedError as e:
    if str(e) == 'beam:transform:test_stream:v1':
        raise RuntimeError('Runner does not support TestStream; use FnApiRunner')

Prevention

When it happens

Trigger: A pipeline contains a TestStream transform (used in streaming tests) and is run with check_requirements against a runner that did not declare common_urns.primitives.TEST_STREAM.urn in supported_requirements.

Common situations: Running Beam streaming unit tests with TestStream on a runner that doesn't support it (e.g. certain portable/foreign-language runners); using TestStream in integration rather than unit test contexts.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/runner.py:218

  def check_requirements(
      self,
      pipeline_proto: beam_runner_api_pb2.Pipeline,
      supported_requirements: Iterable[str]):
    """Check that this runner can satisfy all pipeline requirements."""

    # Imported here to avoid circular dependencies.
    # pylint: disable=wrong-import-order, wrong-import-position
    from apache_beam.runners.portability.fn_api_runner import translations
    supported_requirements = set(supported_requirements)
    for requirement in pipeline_proto.requirements:
      if requirement not in supported_requirements:
        raise ValueError(
            'Unable to run pipeline with requirement: %s' % requirement)
    for transform in pipeline_proto.components.transforms.values():
      if transform.spec.urn == common_urns.primitives.TEST_STREAM.urn:
        if common_urns.primitives.TEST_STREAM.urn not in supported_requirements:
          raise NotImplementedError(transform.spec.urn)
      elif transform.spec.urn in translations.PAR_DO_URNS:
        payload = beam_runner_api_pb2.ParDoPayload.FromString(
            transform.spec.payload)
        for timer in payload.timer_family_specs.values():
          if timer.time_domain not in (
              beam_runner_api_pb2.TimeDomain.EVENT_TIME,
              beam_runner_api_pb2.TimeDomain.PROCESSING_TIME):
            raise NotImplementedError(timer.time_domain)

  def default_pickle_library_override(self):
    """Default pickle library, can be overridden by runner implementation."""
    return None


# FIXME: replace with PipelineState(str, enum.Enum)
class PipelineState(object):
  """State of the Pipeline, as returned by :attr:`PipelineResult.state`.

View on GitHub (pinned to 12126d8942)