{"record":{"id":"631e29d79c595765","repo":"apache/beam","slug":"unconsumed-error-output-for-identify-object-t-the-output","errorCode":null,"errorMessage":"Unconsumed error output for {identify_object(t)}. The output named {transform_name}.{error_output_name} must be used as an input to some other transform. See https://beam.apache.org/documentation/sdks/yaml-errors","messagePattern":"Unconsumed error output for (.+?)\\. The output named (.+?)\\.(.+?) must be used as an input to some other transform\\. See https://beam\\.apache\\.org/documentation/sdks/yaml-errors","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/yaml/yaml_transform.py","lineNumber":1281,"sourceCode":"    consumed = set(\n        scope.get_transform_id_and_output_name(output)\n        for output in spec['output'].values())\n    for t in spec['transforms']:\n      config = t.get('config', t)\n      if 'error_handling' in config:\n        if 'output' not in config['error_handling']:\n          raise ValueError(\n              f'Missing output in error_handling of {identify_object(t)}')\n        to_handle[t['__uuid__'], config['error_handling']['output']] = t\n      for _, input in empty_if_explicitly_empty(t['input']).items():\n        if input not in spec['input']:\n          consumed.add(scope.get_transform_id_and_output_name(input))\n    for error_pcoll, t in to_handle.items():\n      if error_pcoll not in consumed:\n        config = t.get('config', t)\n        transform_name = t.get('name', t.get('type'))\n        error_output_name = config['error_handling']['output']\n        raise ValueError(\n            f'Unconsumed error output for {identify_object(t)}. '\n            f'The output named {transform_name}.{error_output_name} '\n            'must be used as an input to some other transform. '\n            'See https://beam.apache.org/documentation/sdks/yaml-errors')\n  return spec\n\n\ndef lift_config(spec):\n  if 'config' not in spec:\n    common_params = 'name', 'type', 'input', 'output', 'transforms'\n    return {\n        'config': {\n            k: v\n            for (k, v) in spec.items() if k not in common_params\n        },\n        **{\n            k: v\n            for (k, v) in spec.items()  #","sourceCodeStart":1263,"sourceCodeEnd":1299,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/yaml/yaml_transform.py#L1263-L1299","documentation":"Beam YAML requires that error outputs declared via error_handling.output actually be consumed — the error PCollection must feed some other transform. ensure_errors_consumed tracks (transform_id, output_name) pairs and raises this ValueError if a declared error output is never used as an input anywhere, preventing silent data loss of failed records.","triggerScenarios":"ensure_errors_consumed finds a (error_pcoll, t) pair in to_handle where error_pcoll is not in the consumed set — i.e. no transform's input references <that transform>.<error_output_name> and it is not a pipeline output.","commonSituations":"Adding error_handling to a transform but forgetting to wire the error output to a sink; deleting the error-handling branch during refactoring; typo in the output reference so the consumption check misses it.","solutions":["Add a transform whose input is <transform_name>.<error_output_name>, e.g. a WriteToJson or LogForTesting sink for errors.","Route the error output into a normalization/repair transform before sinking it.","Remove error_handling if the error branch is not actually wanted (errors then fail the pipeline instead)."],"exampleFix":"# before\n- type: MapToFields\n  name: parse\n  config:\n    error_handling:\n      output: errors\n# after\n- type: MapToFields\n  name: parse\n  config:\n    error_handling:\n      output: errors\n- type: WriteToJson\n  input: {in: parse.errors}\n  config:\n    path: /tmp/errors.json","handlingStrategy":"validation","validationCode":"consumed = set()\nfor t in spec.get('transforms', []):\n    for inp in t.get('input', {}).values():\n        consumed.add(inp)\nfor t in spec.get('transforms', []):\n    eh = t.get('config', t).get('error_handling')\n    if eh:\n        name = t.get('name', t['type'])\n        if f'{name}.{eh[\"output\"]}' not in consumed:\n            print(f'warning: error output {name}.{eh[\"output\"]} is unconsumed')","typeGuard":"def error_output_consumed(spec, t):\n    eh = t.get('config', t).get('error_handling')\n    if not eh:\n        return True\n    ref = f\"{t.get('name', t['type'])}.{eh['output']}\"\n    return any(ref in i.values() for i in [tr.get('input', {}) for tr in spec['transforms']])","tryCatchPattern":"try:\n    spec = ensure_errors_consumed(spec)\nexcept ValueError as e:\n    if 'Unconsumed error output' in str(e):\n        raise SystemExit(f'{e} — wire the error output into a sink or remove error_handling')\n    raise","preventionTips":["Every error_handling.output must be referenced by some transform's input","Grep the spec for each error output name before running","See https://beam.apache.org/documentation/sdks/yaml-errors for the pattern"],"tags":["python","apache-beam","yaml","error-handling","data-loss"],"backgroundTag":"unconsumed-error-output","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"}