This is an automated email from the ASF dual-hosted git repository.
tvalentyn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 7bc2fb49f8d Fix a memory leak in the failed instruction id cache.
(#39405)
7bc2fb49f8d is described below
commit 7bc2fb49f8dcadb035b4a6caad3da4a6e6f00ab8
Author: tvalentyn <[email protected]>
AuthorDate: Tue Jul 21 07:54:39 2026 -0700
Fix a memory leak in the failed instruction id cache. (#39405)
* Sanitize the exception message stored for failed instructions to avoid
inadvertently capturing the content of stack frames in the heap and limit RAM
growth due to cache size.
* Use Py3.10 syntax
---
.../apache_beam/runners/worker/sdk_worker.py | 12 +++++--
.../apache_beam/runners/worker/sdk_worker_test.py | 38 ++++++++++++++++++++++
2 files changed, 48 insertions(+), 2 deletions(-)
diff --git a/sdks/python/apache_beam/runners/worker/sdk_worker.py
b/sdks/python/apache_beam/runners/worker/sdk_worker.py
index a79cb0e8de6..db25a40a405 100644
--- a/sdks/python/apache_beam/runners/worker/sdk_worker.py
+++ b/sdks/python/apache_beam/runners/worker/sdk_worker.py
@@ -80,7 +80,7 @@ DEFAULT_BUNDLE_PROCESSOR_CACHE_SHUTDOWN_THRESHOLD_S = 60
MAX_KNOWN_NOT_RUNNING_INSTRUCTIONS = 1000
# The number of ProcessBundleRequest instruction ids that BundleProcessorCache
# will remember for failed instructions.
-MAX_FAILED_INSTRUCTIONS = 10000
+MAX_FAILED_INSTRUCTIONS = 1000
# retry on transient UNAVAILABLE grpc error from state channels.
_GRPC_SERVICE_CONFIG = json.dumps({
@@ -559,7 +559,15 @@ class BundleProcessorCache(object):
"""
processor = None
with self._lock:
- self.failed_instruction_ids[instruction_id] = exception
+ tb_str = "".join(traceback.format_exception(exception))
+ if len(tb_str) > 10240:
+ tb_str = (
+ tb_str[:5000] + "\n... [traceback truncated] ...\n" +
+ tb_str[-5000:])
+ clean_exception = RuntimeError(
+ f"Original Exception: {type(exception).__name__}:
{str(exception)[:2000]}\n{tb_str}"
+ )
+ self.failed_instruction_ids[instruction_id] = clean_exception
while len(self.failed_instruction_ids) > MAX_FAILED_INSTRUCTIONS:
self.failed_instruction_ids.popitem(last=False)
if instruction_id in self.active_bundle_processors:
diff --git a/sdks/python/apache_beam/runners/worker/sdk_worker_test.py
b/sdks/python/apache_beam/runners/worker/sdk_worker_test.py
index 76e428f0646..bea313a4d2f 100644
--- a/sdks/python/apache_beam/runners/worker/sdk_worker_test.py
+++ b/sdks/python/apache_beam/runners/worker/sdk_worker_test.py
@@ -296,6 +296,44 @@ class SdkWorkerTest(unittest.TestCase):
worker.do_instruction(split_request).error,
hc.contains_string('test message'))
+ def test_failed_instruction_id_cache_size_is_capped(self):
+ data_channel_factory = mock.create_autospec(
+ data_plane.GrpcClientDataChannelFactory)
+ bundle_processor_cache = BundleProcessorCache(
+ None, None, data_channel_factory, {})
+ if bundle_processor_cache.periodic_shutdown:
+ bundle_processor_cache.periodic_shutdown.cancel()
+
+ with mock.patch(
+ 'apache_beam.runners.worker.sdk_worker.MAX_FAILED_INSTRUCTIONS', 10):
+ for i in range(15):
+ bundle_processor_cache.discard(f'inst_{i}', RuntimeError(f'error {i}'))
+
+ for i in range(5):
+ self.assertNotIn(
+ f'inst_{i}', bundle_processor_cache.failed_instruction_ids)
+ for i in range(5, 15):
+ self.assertIn(
+ f'inst_{i}', bundle_processor_cache.failed_instruction_ids)
+
+ def test_failed_instruction_tracebacks_are_truncated_when_too_long(self):
+ data_channel_factory = mock.create_autospec(
+ data_plane.GrpcClientDataChannelFactory)
+ bundle_processor_cache = BundleProcessorCache(
+ None, None, data_channel_factory, {})
+ if bundle_processor_cache.periodic_shutdown:
+ bundle_processor_cache.periodic_shutdown.cancel()
+
+ long_message = "x" * 15000
+ bundle_processor_cache.discard('instruction_id',
RuntimeError(long_message))
+
+ stored_exception = bundle_processor_cache.failed_instruction_ids[
+ 'instruction_id']
+ tb_str = str(stored_exception)
+
+ self.assertLessEqual(len(tb_str), 13000)
+ self.assertIn('[traceback truncated]', tb_str)
+
def test_data_sampling_response(self):
# Create a data sampler with some fake sampled data. This data will be seen
# in the sample response.
