From 875ff82ccd936e5008b194c6c596fc5ea27dea9e Mon Sep 17 00:00:00 2001 From: Danny Mccormick Date: Thu, 4 Sep 2025 10:04:58 -0400 Subject: [PATCH] Correctly exclude prism when using external transforms --- .../apache_beam/runners/direct/direct_runner.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/sdks/python/apache_beam/runners/direct/direct_runner.py b/sdks/python/apache_beam/runners/direct/direct_runner.py index 0af0ca8d3175..487d2a8cbe25 100644 --- a/sdks/python/apache_beam/runners/direct/direct_runner.py +++ b/sdks/python/apache_beam/runners/direct/direct_runner.py @@ -137,6 +137,14 @@ def accept(self, pipeline, is_interactive): self.supported_by_prism_runner = False else: pipeline.visit(self) + # Avoid circular import + from apache_beam.pipeline import ExternalTransformFinder + if ExternalTransformFinder.contains_external_transforms(pipeline): + # TODO(https://github.com/apache/beam/issues/33623): Prism currently + # seems to not be able to consistently bring up external transforms. + # It does sometimes, but at volume suites start to fail. We will try + # to enable this in a future release. + self.supported_by_prism_runner = False return self.supported_by_prism_runner def visit_transform(self, applied_ptransform): @@ -145,12 +153,6 @@ def visit_transform(self, applied_ptransform): # being used. if isinstance(transform, TestStream): self.supported_by_prism_runner = False - if isinstance(transform, beam.ExternalTransform): - # TODO(https://github.com/apache/beam/issues/33623): Prism currently - # seems to not be able to consistently bring up external transforms. - # It does sometimes, but at volume suites start to fail. We will try - # to enable this in a future release. - self.supported_by_prism_runner = False if isinstance(transform, beam.ParDo): dofn = transform.dofn # TODO(https://github.com/apache/beam/issues/33623): Prism currently