{"record":{"id":"b8b35505c25ec18c","repo":"apache/beam","slug":"unexpected-type-for-bad-results-type-bad-results","errorCode":null,"errorMessage":"Unexpected type for bad_results: {type(bad_results)}","messagePattern":"Unexpected type for bad_results: (.+?)","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/ml/transforms/base.py","lineNumber":437,"sourceCode":"      for i in range(len(ptransform_list)):\n        if hasattr(ptransform_list[i], 'artifact_mode'):\n          ptransform_list[i].artifact_mode = self._artifact_mode\n\n    transform_name = None\n    for ptransform in ptransform_list:\n      if self._with_exception_handling:\n        if hasattr(ptransform, 'with_exception_handling'):\n          ptransform = ptransform.with_exception_handling(\n              **self._exception_handling_args)\n          pcoll, bad_results = pcoll | ptransform\n          # RunInference outputs a RunInferenceDLQ instead of a PCollection.\n          # since TFTProcessHandler and RunInferene are supported, try to infer\n          # the type of bad_results and append it to the list of errors.\n          if isinstance(bad_results, RunInferenceDLQ):\n            bad_results = bad_results.failed_inferences\n            transform_name = ptransform.annotations()['model_handler']\n          elif not isinstance(bad_results, beam.PCollection):\n            raise NotImplementedError(\n                f'Unexpected type for bad_results: {type(bad_results)}')\n          bad_results = bad_results | beam.Map(\n              lambda x: _map_errors_to_beam_row(x, transform_name))\n          upstream_errors.append(bad_results)\n\n      else:\n        pcoll = pcoll | ptransform\n    _ = (\n        pcoll.pipeline\n        | \"MLTransformMetricsUsage\" >> MLTransformMetricsUsage(self))\n\n    if self._with_exception_handling:\n      bad_pcoll = (upstream_errors | beam.Flatten())\n      return pcoll, bad_pcoll  # type: ignore[return-value]\n    return pcoll  # type: ignore[return-value]\n\n  def with_transform(self, transform: MLTransformProvider):\n    \"\"\"","sourceCodeStart":419,"sourceCodeEnd":455,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/ml/transforms/base.py#L419-L455","documentation":"When MLTransform's exception handling collects failed results, bad_results must be a RunInferenceDLQ or a beam.PCollection; any other type cannot be mapped to error rows, so NotImplementedError is raised naming the unexpected type.","triggerScenarios":"Using with_exception_handling and supplying a bad_results sink/collector that is neither RunInferenceDLQ nor a PCollection (e.g. a plain list, callable, or custom DLQ object).","commonSituations":"Wiring a custom dead-letter container class instead of a PCollection; passing a function that returns a PCollection rather than the PCollection itself.","solutions":["Pass a beam.PCollection of failed records as bad_results.","For RunInference failures, pass the RunInferenceDLQ object returned by the model handler setup.","Convert custom error containers to a PCollection first (beam.Create(...) | ...)."],"exampleFix":"// before\nmltransform.with_exception_handling(bad_results=my_error_list)\n// after\nbad = failed_records | beam.Map(lambda x: ...)  # a PCollection\nmltransform.with_exception_handling(bad_results=bad)","handlingStrategy":"type-guard","validationCode":"from apache_beam.ml.inference.base import RunInferenceDLQ\nassert isinstance(bad_results, (RunInferenceDLQ, __import__('apache_beam', fromlist=['PCollection']).PCollection)), f'bad bad_results type: {type(bad_results)}'","typeGuard":"def is_valid_bad_results(br) -> bool:\n    import apache_beam as beam\n    from apache_beam.ml.inference.base import RunInferenceDLQ\n    return isinstance(br, (RunInferenceDLQ, beam.PCollection))","tryCatchPattern":"try:\n    out = mltransform.with_exception_handling(bad_results=br)\nexcept NotImplementedError as e:\n    if 'Unexpected type for bad_results' in str(e):\n        br = failures | beam.Map(lambda x: x)\n        out = mltransform.with_exception_handling(bad_results=br)","preventionTips":["Only pass PCollections or RunInferenceDLQ as bad_results","Unwrap custom DLQ wrappers to their underlying PCollection before wiring","Type-hint bad_results in your pipeline helpers"],"tags":["python","apache-beam","type-mismatch"],"backgroundTag":"type-mismatch","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"}