{"record":{"id":"df185cd77e8fb3e1","repo":"apache/beam","slug":"a-cluster-identifier-should-be-optional-union-str-beam","errorCode":null,"errorMessage":"A cluster_identifier should be Optional[Union[str, beam.Pipeline, ClusterMetadata], instead %s was given.","messagePattern":"A cluster_identifier should be Optional\\[Union\\[str, beam\\.Pipeline, ClusterMetadata\\], instead (.+?) was given\\.","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/interactive_beam.py","lineNumber":594,"sourceCode":"      elif isinstance(cluster_identifier, ClusterMetadata):\n        meta = cluster_identifier\n        if meta in self.dataproc_cluster_managers:\n          meta = self.dataproc_cluster_managers[meta].cluster_metadata\n        elif (meta and self.default_cluster_metadata and\n              meta.cluster_name == self.default_cluster_metadata.cluster_name):\n          _LOGGER.warning(\n              'Cannot change the configuration of the running cluster %s. '\n              'Existing is %s, desired is %s.',\n              self.default_cluster_metadata.cluster_name,\n              self.default_cluster_metadata,\n              meta)\n          meta.reset_name()\n          _LOGGER.warning(\n              'To avoid conflict, issuing a new cluster name %s '\n              'for a new cluster.',\n              meta.cluster_name)\n      else:\n        raise TypeError(\n            'A cluster_identifier should be Optional[Union[str, '\n            'beam.Pipeline, ClusterMetadata], instead %s was given.',\n            type(cluster_identifier))\n    return meta\n\n  def _cleanup(self, dcm: DataprocClusterManager) -> None:\n    dcm.cleanup()\n    self.dataproc_cluster_managers.pop(dcm.cluster_metadata, None)\n    self.master_urls.pop(dcm.cluster_metadata.master_url, None)\n    for p in dcm.pipelines:\n      self.pipelines.pop(p, None)\n    if dcm.cluster_metadata == self.default_cluster_metadata:\n      self.default_cluster_metadata = None\n\n\n# Users can set options to guide how Interactive Beam works.\n# Examples:\n# ib.options.enable_recording_replay = False/True","sourceCodeStart":576,"sourceCodeEnd":612,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/interactive_beam.py#L576-L612","documentation":"Interactive Beam's cluster manager helper (_cluster_metadata) resolves a cluster_identifier into a ClusterMetadata. It only accepts a cluster name string, a beam.Pipeline, or an existing ClusterMetadata; anything else is rejected with a TypeError raised from the DataprocClusterManager machinery in interactive_beam.py. Note: the raise passes the message and argument as tuple positionals, so the printed message ends with a raw tuple rather than a formatted string, but the intent is clear: the cluster_identifier you passed is of an unsupported type.","triggerScenarios":"Calling ib.cluster_metadata (and related evict/show_cluster_into paths) with a cluster_identifier that is not a str, not a beam.Pipeline, and not a ClusterMetadata — e.g. passing an int pipeline id, a PipelineResult, a dict of options, or a wrapper object holding the pipeline.","commonSituations":"Notebook users wiring up interactive Dataproc workflows: storing a pipeline handle in a wrapper and passing the wrapper; passing the result of pipeline.run() (a PipelineResult) instead of the pipeline; typing a cluster id into a variable holding bytes or a numpy str_; older notebooks written against an API version that accepted other identifier shapes.","solutions":["Pass a ClusterMetadata object obtained from ib.cluster_metadata instead of a raw handle.","Pass the beam.Pipeline object itself (pipeline = beam.Pipeline(); ...; ib.cluster_metadata(pipeline)).","Pass the cluster name string exactly as returned by the cluster manager (meta.cluster_name).","If you must accept arbitrary input, coerce to one of the three supported types before calling.","Read the printed tuple in the traceback to see the actual type you passed (the second tuple element is type(cluster_identifier)) and fix the variable's source."],"exampleFix":"// before: result = pipeline.run(); ib.cluster_metadata(result)  # PipelineResult, not accepted | // after: ib.cluster_metadata(pipeline)  # pass the beam.Pipeline itself, a str name, or a ClusterMetadata","handlingStrategy":"type-guard","validationCode":"import apache_beam as beam; from apache_beam.runners.interactive.dataproc.types import ClusterMetadata; def is_valid_cluster_identifier(x): return x is None or isinstance(x, (str, beam.Pipeline, ClusterMetadata)); assert is_valid_cluster_identifier(ident)","typeGuard":"def as_cluster_identifier(x): import apache_beam as beam; from apache_beam.runners.interactive.dataproc.types import ClusterMetadata; return x if x is None or isinstance(x, (str, beam.Pipeline, ClusterMetadata)) else (_ for _ in ()).throw(TypeError(f'Unsupported cluster_identifier type: {type(x)}'))","tryCatchPattern":"try: ib.cluster_metadata(ident) | except TypeError as e: logger.error('Bad cluster_identifier %r: %s', ident, e); ident = ib.cluster_metadata(pipeline)  # fall back to pipeline handle","preventionTips":["Only pass values returned by ib.cluster_metadata, the pipeline object, or meta.cluster_name.","Normalize notebook UI inputs to one of the three supported types at the call site.","Remember PipelineResult is NOT accepted — keep the pipeline object around.","Check type(cluster_identifier) before calling when wiring wrappers."],"tags":["python","apache-beam","interactive","type-error","dataproc"],"backgroundTag":"invalid-argument-value","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}