diff --git a/.github/workflows/build_wheels.yml b/.github/workflows/build_wheels.yml
index 6236888521a5..02b223943fea 100644
--- a/.github/workflows/build_wheels.yml
+++ b/.github/workflows/build_wheels.yml
@@ -37,7 +37,8 @@ concurrency:
   cancel-in-progress: true
 
 env:
-  GCP_PATH: "gs://${{ secrets.GCP_PYTHON_WHEELS_BUCKET }}/${GITHUB_REF##*/}/${GITHUB_SHA}-${GITHUB_RUN_ID}/"
+  # Use github.* context; workflow env is not bash-evaluated.
+  GCP_PATH: "gs://${{ secrets.GCP_PYTHON_WHEELS_BUCKET }}/${{ github.ref_name }}/${{ github.sha }}-${{ github.run_id }}/"
 
 
 jobs:
diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml
index be534203eaed..7072b9a8da4a 100644
--- a/.github/workflows/codeql.yml
+++ b/.github/workflows/codeql.yml
@@ -36,7 +36,7 @@ jobs:
       matrix: ${{ steps.set-matrix.outputs.matrix }}
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v4
+        uses: actions/checkout@v7
       - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d
         id: filter
         with:
diff --git a/CHANGES.md b/CHANGES.md
index d39a62f1a780..991c341460a0 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -84,7 +84,7 @@
 
 * Fixed unbounded checkpoint state growth for splittable DoFns that self-checkpoint on the portable Flink runner (Java) ([#27648](https://github.com/apache/beam/issues/27648)).
 * Improved Java pipeline performance by avoiding repeated `DoFn` type descriptor resolution when creating cached invokers ([#39309](https://github.com/apache/beam/issues/39309)).
-* Fixed X (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* (Python) Fixed a memory leak in Python SDK caused by storing exceptions with potentially large stack frames in a cache ([#39406](https://github.com/apache/beam/issues/39406)).
 
 ## Security Fixes
 
@@ -138,6 +138,7 @@
 ## Known Issues
 
 * (Java) Projects using the Flink runner with Flink 2.1 or later alongside libraries requiring `org.lz4:lz4-java` (e.g., Kafka clients) may encounter a Gradle capability conflict, because Flink 2.1+ ships `at.yawk.lz4:lz4-java` which declares the same capability. To resolve, add a `capabilitiesResolution` rule to your `build.gradle` that selects `at.yawk.lz4:lz4-java` ([#38947](https://github.com/apache/beam/issues/38947)).
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
 
 # [2.74.0] - 2026-06-02
 
@@ -179,6 +180,10 @@
 * Fixed BigQueryEnrichmentHandler batch mode dropping earlier requests when multiple requests share the same enrichment key (Python) ([#38035](https://github.com/apache/beam/issues/38035)).
 * Added `max_batch_duration_secs` passthrough support in Python Enrichment BigQuery and CloudSQL handlers so batching duration can be forwarded to `BatchElements` ([#38243](https://github.com/apache/beam/issues/38243)).
 
+## Known Issues
+
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
+
 # [2.73.0] - 2026-04-29
 
 ## Highlights
@@ -215,6 +220,10 @@
 * Fixed [CVE-2023-46604](https://www.cve.org/CVERecord?id=CVE-2023-46604) (CVSS 10.0) and [CVE-2022-41678](https://www.cve.org/CVERecord?id=CVE-2022-41678) by upgrading ActiveMQ from 5.14.5 to 5.19.2 (Java) ([#37943](https://github.com/apache/beam/issues/37943)).
 * Fixed [CVE-2024-1597](https://www.cve.org/CVERecord?id=CVE-2024-1597), [CVE-2022-31197](https://www.cve.org/CVERecord?id=CVE-2022-31197), and [CVE-2022-21724](https://www.cve.org/CVERecord?id=CVE-2022-21724) by upgrading PostgreSQL JDBC Driver from 42.2.16 to 42.6.2 (Java) ([#37942](https://github.com/apache/beam/issues/37942)).
 
+## Known Issues
+
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
+
 # [2.72.0] - 2026-03-30
 
 ## Highlights
@@ -248,6 +257,10 @@
 
 * Fixed [CVE-2024-28397](https://www.cve.org/CVERecord?id=CVE-2024-28397) by switching from js2py to pythonmonkey (Yaml) ([#37560](https://github.com/apache/beam/issues/37560)).
 
+## Known Issues
+
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
+
 # [2.71.0] - 2026-01-22
 
 ## I/Os
@@ -269,6 +282,7 @@
 
 ## Known Issues
 
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
 
 # [2.70.0] - 2025-12-16
 
@@ -291,6 +305,10 @@ Now Beam has full support for Milvus integration including Milvus enrichment and
 
 * (Python) Python 3.9 reached EOL in October 2025 and support for the language version has been removed. ([#36665](https://github.com/apache/beam/issues/36665)).
 
+## Known Issues
+
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
+
 # [2.69.0] - 2025-10-28
 
 ## Highlights
@@ -350,6 +368,10 @@ Now Beam has full support for Milvus integration including Milvus enrichment and
   ([#36141](https://github.com/apache/beam/issues/36141)).
 * Fixed Spanner Change Stream reading stuck issue due to watermark of partition moving backwards ([#36470](https://github.com/apache/beam/issues/36470)).
 
+## Known Issues
+
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
+
 # [2.68.0] - 2025-09-22
 
 ## Highlights
@@ -404,6 +426,7 @@ Now Beam has full support for Milvus integration including Milvus enrichment and
 ## Known Issues
 
 * ([#36470](https://github.com/apache/beam/issues/36470)). Spanner Change Stream reading stuck issue due to watermark of partition moving backwards. This issue exists in 2.67.0 and 2.68.0. To mitigate the issue, either use old version 2.66.0 or go to 2.69.0.
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
 
 # [2.67.0] - 2025-08-12
 
@@ -452,6 +475,7 @@ Now Beam has full support for Milvus integration including Milvus enrichment and
 
 * ([#35666](https://github.com/apache/beam/issues/35666)). YAML Flatten incorrectly drops fields when input PCollections' schema are different. This issue exists for all versions since 2.52.0.
 * ([#36470](https://github.com/apache/beam/issues/36470)). Spanner Change Stream reading stuck issue due to watermark of partition moving backwards. This issue exists in 2.67.0 and 2.68.0. To mitigate the issue, either use old version 2.66.0 or go to 2.69.0.
+* (Python) Long-running Python pipelines might experience memory growth and periodic OOMs ([#39406](https://github.com/apache/beam/issues/39406)).
 
 # [2.66.0] - 2025-07-01
 
diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index d73f2e7a2bae..70a22e397724 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -613,7 +613,7 @@ class BeamModulePlugin implements Plugin<Project> {
     def autoservice_version = "1.0.1"
     def aws_java_sdk2_version = "2.20.162"
     def cassandra_driver_version = "3.10.2"
-    def cdap_version = "6.5.1"
+    def cdap_version = "6.11.4"
     def checkerframework_version = "3.42.0"
     def classgraph_version = "4.8.162"
     def delta_lake_version = "4.2.0"
diff --git a/examples/notebooks/beam-ml/rag_usecase/opensearch_connector.py b/examples/notebooks/beam-ml/rag_usecase/opensearch_connector.py
index fc83c8d443cc..e9f07daef101 100644
--- a/examples/notebooks/beam-ml/rag_usecase/opensearch_connector.py
+++ b/examples/notebooks/beam-ml/rag_usecase/opensearch_connector.py
@@ -88,8 +88,8 @@ def __init__(self,
         """
         self.host = host
         self.port = port
-        self.username = username | os.getenv("OPENSEARCH_USERNAME")
-        self.password = password | os.getenv("OPENSEARCH_PASSWORD")
+        self.username = username or os.getenv("OPENSEARCH_USERNAME")
+        self.password = password or os.getenv("OPENSEARCH_PASSWORD")
         self._batch_size = batch_size
 
         if not self.username or not self.password:
@@ -247,8 +247,8 @@ def __init__(self,
         """
         self.host = host
         self.port = port
-        self.username = username | os.getenv("OPENSEARCH_USERNAME")
-        self.password = password | os.getenv("OPENSEARCH_PASSWORD")
+        self.username = username or os.getenv("OPENSEARCH_USERNAME")
+        self.password = password or os.getenv("OPENSEARCH_PASSWORD")
         self.batch_size = batch_size
         self.embedded_columns = embedded_columns
 
diff --git a/examples/notebooks/beam-ml/rag_usecase/opensearch_enrichment.py b/examples/notebooks/beam-ml/rag_usecase/opensearch_enrichment.py
index 70397550241f..c43d9af181ba 100644
--- a/examples/notebooks/beam-ml/rag_usecase/opensearch_enrichment.py
+++ b/examples/notebooks/beam-ml/rag_usecase/opensearch_enrichment.py
@@ -77,8 +77,8 @@ def __init__(
         """
         self.opensearch_host = opensearch_host
         self.opensearch_port = opensearch_port
-        self.username = username | os.getenv("OPENSEARCH_USERNAME")
-        self.password = password | os.getenv("OPENSEARCH_PASSWORD")
+        self.username = username or os.getenv("OPENSEARCH_USERNAME")
+        self.password = password or os.getenv("OPENSEARCH_PASSWORD")
         self.index_name = index_name
         self.vector_field = vector_field
         self.k = k
diff --git a/sdks/java/io/cdap/build.gradle b/sdks/java/io/cdap/build.gradle
index e0af1717e162..cc7e3ba12cde 100644
--- a/sdks/java/io/cdap/build.gradle
+++ b/sdks/java/io/cdap/build.gradle
@@ -90,3 +90,8 @@ test {
     // Open java.lang for Gson reflection on StackTraceElement under Java 17+
     jvmArgs '--add-opens=java.base/java.lang=ALL-UNNAMED'
 }
+
+// spark3_streaming depends on old 'org.lz4:lz4-java'; conflict with kafka-clients:3.9.2
+configurations.all {
+    resolveCapabilitiesConflict(it, 'org.lz4:lz4-java', 'at.yawk.lz4')
+}
\ No newline at end of file
diff --git a/sdks/python/apache_beam/examples/ml-orchestration/kfp/components/train/requirements.txt b/sdks/python/apache_beam/examples/ml-orchestration/kfp/components/train/requirements.txt
index 5b996fdfc70a..2fc217218387 100644
--- a/sdks/python/apache_beam/examples/ml-orchestration/kfp/components/train/requirements.txt
+++ b/sdks/python/apache_beam/examples/ml-orchestration/kfp/components/train/requirements.txt
@@ -15,4 +15,4 @@
 
 torch==2.12.0
 numpy==1.22.4
-Pillow==10.2.0
\ No newline at end of file
+Pillow==12.3.0
\ No newline at end of file
diff --git a/sdks/python/apache_beam/runners/worker/sdk_worker.py b/sdks/python/apache_beam/runners/worker/sdk_worker.py
index a79cb0e8de6e..db25a40a405c 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 @@
 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 @@ def discard(self, instruction_id, exception):
     """
     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 76e428f06464..bea313a4d2fd 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 @@ def test_failed_bundle_processor_returns_failed_split_response(self):
         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.