apache/beam · error · ValueError

Missing requirement declaration

Error message

Missing requirement declaration: %s

What it means

_validate_requirements scans the pipeline proto for transforms that imply runner requirements (e.g. requires_bundle_finalization, requires_stateful_processing) and verifies the pipeline declares each of them in pipeline.requirements. A missing declaration means the graph was produced by a producer that failed to register a requirement it needs.

Solutions

1. Rebuild/serialize the pipeline with a matching apache-beam version so requirements are declared
2. Add the missing requirement strings to pipeline_proto.requirements when constructing the proto manually
3. Check for version skew between the pipeline producer and the FnApiRunner
4. Update any custom translations that add transforms without adding their requirements

Example fix

// before
pipeline_proto.components.transforms[...].spec.urn = STATEFUL_DOFN_URN  # requirement not declared
// after
pipeline_proto.requirements.append(common_urns.requirements.REQUIRES_STATEFUL_PROCESSING.urn)

expected = collect_expected_requirements(pipeline_proto)
missing = expected - set(pipeline_proto.requirements)
assert not missing, missing

def requirements_declared(pipeline_proto, expected) -> bool:
    return expected.issubset(set(pipeline_proto.requirements))

try:
    runner.run_via_runner_api(proto)
except ValueError as e:
    if 'Missing requirement declaration' in str(e):
        proto = rebuild_proto_with_requirements(proto)
    raise

When it happens

Trigger: Running a pipeline whose transforms (stateful DoFns, timers, finalization) require capabilities not listed in pipeline_proto.requirements — usually a hand-crafted or stale pipeline proto, or a runner-api translation bug.

Common situations: Custom pipeline construction via the runner API; mismatches when a pipeline serialized by an older Beam version is executed by a newer runner; direct-arg (no PickledRunner) pipelines.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• An unsupported sink was specified
• At least one of --render_port or --render_output must be…
• buffer_sec must be >= 0, got
• Cannot skip negative number of header lines
• change_function must be 'CHANGES' or 'APPENDS', got

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py:352

          expected_requirements.add(
              common_urns.requirements.REQUIRES_STATEFUL_PROCESSING.urn)
        if payload.requires_stable_input:
          expected_requirements.add(
              common_urns.requirements.REQUIRES_STABLE_INPUT.urn)
        if payload.requires_time_sorted_input:
          expected_requirements.add(
              common_urns.requirements.REQUIRES_TIME_SORTED_INPUT.urn)
        if payload.restriction_coder_id:
          expected_requirements.add(
              common_urns.requirements.REQUIRES_SPLITTABLE_DOFN.urn)
      else:
        for sub in transform.subtransforms:
          add_requirements(sub)

    for root in pipeline_proto.root_transform_ids:
      add_requirements(root)
    if not expected_requirements.issubset(pipeline_proto.requirements):
      raise ValueError(
          'Missing requirement declaration: %s' %
          (expected_requirements - set(pipeline_proto.requirements)))

  def _check_requirements(
      self, pipeline_proto: beam_runner_api_pb2.Pipeline) -> None:
    """Check that this runner can satisfy all pipeline requirements."""
    supported_requirements = set(self.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:
        raise NotImplementedError(transform.spec.urn)
      elif transform.spec.urn in translations.PAR_DO_URNS:
        payload = proto_utils.parse_Bytes(
            transform.spec.payload, beam_runner_api_pb2.ParDoPayload)
        for timer in payload.timer_family_specs.values():

View on GitHub (pinned to 12126d8942)