From 4fd48243107763951d11642fd4b31545767fdcef Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Thu, 18 Sep 2025 13:08:52 -0700 Subject: [PATCH 1/5] Timeout execution tree creation for SDK worker ops. --- .../runners/worker/bundle_processor.py | 22 ++++++++++++++++++- .../runners/worker/worker_status.py | 16 +++++++++++--- 2 files changed, 34 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/runners/worker/bundle_processor.py b/sdks/python/apache_beam/runners/worker/bundle_processor.py index ad48358d588e..3b018ea6a182 100644 --- a/sdks/python/apache_beam/runners/worker/bundle_processor.py +++ b/sdks/python/apache_beam/runners/worker/bundle_processor.py @@ -24,6 +24,7 @@ import base64 import bisect import collections +import concurrent.futures import copy import heapq import itertools @@ -76,6 +77,7 @@ from apache_beam.runners.worker import operation_specs from apache_beam.runners.worker import operations from apache_beam.runners.worker import statesampler +from apache_beam.runners.worker.worker_status import thread_dump from apache_beam.transforms import TimeDomain from apache_beam.transforms import core from apache_beam.transforms import environments @@ -1130,7 +1132,25 @@ def __init__( 'fnapi-step-%s' % self.process_bundle_descriptor.id, self.counter_factory) - self.ops = self.create_execution_tree(self.process_bundle_descriptor) + with concurrent.futures.ThreadPoolExecutor( + max_workers=1, thread_name_prefix='ExecutionTreeCreator') as executor: + future = executor.submit( + self.create_execution_tree, self.process_bundle_descriptor) + try: + self.ops = future.result(timeout=3600) + except concurrent.futures.TimeoutError: + # In rare cases, unpickling a DoFn might get permanently stuck, + # for example when unpickling involves importing a module and + # a subprocess is launched during the import operation. + _LOGGER.error( + 'Timed out when creating execution tree for %s.\n%s', + self.process_bundle_descriptor.id, + thread_dump('ExecutionTreeCreator')) + # Raising an exception here doesn't interrupt the left-over thread. + # Out of caution, terminate the SDK harness process. + from apache_beam.runners.worker.sdk_worker_main import terminate_sdk_harness + terminate_sdk_harness() + for op in reversed(self.ops.values()): op.setup(self.data_sampler) self.splitting_lock = threading.Lock() diff --git a/sdks/python/apache_beam/runners/worker/worker_status.py b/sdks/python/apache_beam/runners/worker/worker_status.py index 86a7b5e8ee1a..1d9e30efe3c7 100644 --- a/sdks/python/apache_beam/runners/worker/worker_status.py +++ b/sdks/python/apache_beam/runners/worker/worker_status.py @@ -66,13 +66,23 @@ def _current_frames(): return sys._current_frames() # pylint: disable=protected-access -def thread_dump(): - """Get a thread dump for the current SDK worker harness. """ +def thread_dump(thread_prefix=None): + """Get a thread dump for the current SDK harness. + + Args: + thread_prefix: (str) An optional prefix to filter threads by. + """ # deduplicate threads with same stack trace stack_traces = defaultdict(list) frames = _current_frames() - for t in threading.enumerate(): + threads_to_dump = threading.enumerate() + if thread_prefix: + threads_to_dump = [ + t for t in threads_to_dump if t.name.startswith(thread_prefix) + ] + + for t in threads_to_dump: try: stack_trace = ''.join(traceback.format_stack(frames[t.ident])) except KeyError: From 8410a06c0965d7e406c60d61f18b877532370292 Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Thu, 18 Sep 2025 17:13:50 -0700 Subject: [PATCH 2/5] Run tests --- .github/trigger_files/beam_PostCommit_Python.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index 1fa29a890c2f..815b511b8988 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 29 + "modification": 30 } From 95ee28287b0351cf3a1d2eba768afc8b4cdb5dcf Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Thu, 25 Sep 2025 15:19:40 -0700 Subject: [PATCH 3/5] Update the error message --- sdks/python/apache_beam/runners/worker/bundle_processor.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/sdks/python/apache_beam/runners/worker/bundle_processor.py b/sdks/python/apache_beam/runners/worker/bundle_processor.py index 3b018ea6a182..afb4a7ab23ad 100644 --- a/sdks/python/apache_beam/runners/worker/bundle_processor.py +++ b/sdks/python/apache_beam/runners/worker/bundle_processor.py @@ -1143,7 +1143,12 @@ def __init__( # for example when unpickling involves importing a module and # a subprocess is launched during the import operation. _LOGGER.error( - 'Timed out when creating execution tree for %s.\n%s', + 'Timed out while reconstructing a pipeline fragment for: %s.\n' + 'Likely, this a rare and transient error. The SDK harness ' + 'will self-terminate, and the runner can retry the operation. ' + 'If the error persists, investigate whether the stuckness happens ' + 'while deserializing (unpickling) a dependency of your pipeline ' + 'in the stacktrace below: \n%s\n', self.process_bundle_descriptor.id, thread_dump('ExecutionTreeCreator')) # Raising an exception here doesn't interrupt the left-over thread. From 8d352f8c90294fd41ff81f55e793c6d6a3feb05b Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Thu, 25 Sep 2025 16:51:44 -0700 Subject: [PATCH 4/5] Minor change to termination logic --- sdks/python/apache_beam/runners/worker/sdk_worker_main.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/sdks/python/apache_beam/runners/worker/sdk_worker_main.py b/sdks/python/apache_beam/runners/worker/sdk_worker_main.py index 7ea0e0eb1099..cdb807e8dbc5 100644 --- a/sdks/python/apache_beam/runners/worker/sdk_worker_main.py +++ b/sdks/python/apache_beam/runners/worker/sdk_worker_main.py @@ -233,6 +233,10 @@ def terminate_sdk_harness(): if _FN_LOG_HANDLER: _FN_LOG_HANDLER.close() os.kill(os.getpid(), signal.SIGINT) + # Delay further control flow in the caller until process is terminated. + time.sleep(60) + # Try to force-terminate if still running. + os.kill(os.getpid(), signal.SIGKILL) def _load_pipeline_options(options_json): From b7b038ea1332c98abfbed91f95727b6e47715eb7 Mon Sep 17 00:00:00 2001 From: Valentyn Tymofieiev Date: Thu, 25 Sep 2025 20:38:16 -0700 Subject: [PATCH 5/5] Grammar --- sdks/python/apache_beam/runners/worker/bundle_processor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/runners/worker/bundle_processor.py b/sdks/python/apache_beam/runners/worker/bundle_processor.py index afb4a7ab23ad..85f1e43d6039 100644 --- a/sdks/python/apache_beam/runners/worker/bundle_processor.py +++ b/sdks/python/apache_beam/runners/worker/bundle_processor.py @@ -1144,9 +1144,9 @@ def __init__( # a subprocess is launched during the import operation. _LOGGER.error( 'Timed out while reconstructing a pipeline fragment for: %s.\n' - 'Likely, this a rare and transient error. The SDK harness ' + 'This is likely a transient error. The SDK harness ' 'will self-terminate, and the runner can retry the operation. ' - 'If the error persists, investigate whether the stuckness happens ' + 'If the error is frequent, check whether the stuckness happens ' 'while deserializing (unpickling) a dependency of your pipeline ' 'in the stacktrace below: \n%s\n', self.process_bundle_descriptor.id,