{"record":{"id":"8913ce5cea4f1755","repo":"apache/beam","slug":"missing-requirement-declaration-s","errorCode":null,"errorMessage":"Missing requirement declaration: %s","messagePattern":"Missing requirement declaration: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py","lineNumber":352,"sourceCode":"          expected_requirements.add(\n              common_urns.requirements.REQUIRES_STATEFUL_PROCESSING.urn)\n        if payload.requires_stable_input:\n          expected_requirements.add(\n              common_urns.requirements.REQUIRES_STABLE_INPUT.urn)\n        if payload.requires_time_sorted_input:\n          expected_requirements.add(\n              common_urns.requirements.REQUIRES_TIME_SORTED_INPUT.urn)\n        if payload.restriction_coder_id:\n          expected_requirements.add(\n              common_urns.requirements.REQUIRES_SPLITTABLE_DOFN.urn)\n      else:\n        for sub in transform.subtransforms:\n          add_requirements(sub)\n\n    for root in pipeline_proto.root_transform_ids:\n      add_requirements(root)\n    if not expected_requirements.issubset(pipeline_proto.requirements):\n      raise ValueError(\n          'Missing requirement declaration: %s' %\n          (expected_requirements - set(pipeline_proto.requirements)))\n\n  def _check_requirements(\n      self, pipeline_proto: beam_runner_api_pb2.Pipeline) -> None:\n    \"\"\"Check that this runner can satisfy all pipeline requirements.\"\"\"\n    supported_requirements = set(self.supported_requirements())\n    for requirement in pipeline_proto.requirements:\n      if requirement not in supported_requirements:\n        raise ValueError(\n            'Unable to run pipeline with requirement: %s' % requirement)\n    for transform in pipeline_proto.components.transforms.values():\n      if transform.spec.urn == common_urns.primitives.TEST_STREAM.urn:\n        raise NotImplementedError(transform.spec.urn)\n      elif transform.spec.urn in translations.PAR_DO_URNS:\n        payload = proto_utils.parse_Bytes(\n            transform.spec.payload, beam_runner_api_pb2.ParDoPayload)\n        for timer in payload.timer_family_specs.values():","sourceCodeStart":334,"sourceCodeEnd":370,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner.py#L334-L370","documentation":"_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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Rebuild/serialize the pipeline with a matching apache-beam version so requirements are declared","Add the missing requirement strings to pipeline_proto.requirements when constructing the proto manually","Check for version skew between the pipeline producer and the FnApiRunner","Update any custom translations that add transforms without adding their requirements"],"exampleFix":"// before\npipeline_proto.components.transforms[...].spec.urn = STATEFUL_DOFN_URN  # requirement not declared\n// after\npipeline_proto.requirements.append(common_urns.requirements.REQUIRES_STATEFUL_PROCESSING.urn)\n","handlingStrategy":"validation","validationCode":"expected = collect_expected_requirements(pipeline_proto)\nmissing = expected - set(pipeline_proto.requirements)\nassert not missing, missing","typeGuard":"def requirements_declared(pipeline_proto, expected) -> bool:\n    return expected.issubset(set(pipeline_proto.requirements))","tryCatchPattern":"try:\n    runner.run_via_runner_api(proto)\nexcept ValueError as e:\n    if 'Missing requirement declaration' in str(e):\n        proto = rebuild_proto_with_requirements(proto)\n    raise","preventionTips":["Build pipeline protos only through the official translations API","Keep producer and runner Beam versions aligned","When adding transforms in custom translations, also add their requirements"],"tags":["python","apache-beam","pipeline-requirements","validation"],"backgroundTag":"missing-required-config-field","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"}