(beam) branch master updated: [Prism] Fix race condition in element manager (#38734)

[shunping]

This is an automated email from the ASF dual-hosted git repository.

shunping 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 a32774afe93 [Prism] Fix race condition in element manager (#38734)
a32774afe93 is described below

commit a32774afe93c451dd065f441cdc1b478f7b6e575
Author: Shunping Huang <shunp...@google.com>
AuthorDate: Thu May 28 17:49:51 2026 -0400

    [Prism] Fix race condition in element manager (#38734)
---
 .github/trigger_files/beam_PostCommit_Python_Versions.json       | 2 +-
 runners/prism/java/build.gradle                                  | 1 +
 sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go | 7 ++++++-
 3 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/.github/trigger_files/beam_PostCommit_Python_Versions.json 
b/.github/trigger_files/beam_PostCommit_Python_Versions.json
index 9cc78c7d1c6..8b2c8c445c1 100644
--- a/.github/trigger_files/beam_PostCommit_Python_Versions.json
+++ b/.github/trigger_files/beam_PostCommit_Python_Versions.json
@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to 
run",
-  "revision": 4
+  "revision": 5
 }
diff --git a/runners/prism/java/build.gradle b/runners/prism/java/build.gradle
index c89974cb6ea..9357515f36c 100644
--- a/runners/prism/java/build.gradle
+++ b/runners/prism/java/build.gradle
@@ -185,6 +185,7 @@ def sickbayTests = [
 def createPrismValidatesRunnerTask = { name, environmentType ->
   Task vrTask = tasks.create(name: name, type: Test, group: "Verification") {
     description "PrismRunner Java $environmentType ValidatesRunner suite"
+    outputs.upToDateWhen { false }
     classpath = configurations.validatesRunner
 
     var prismBuildTask = dependsOn(':runners:prism:build')
diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go 
b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
index f720be20e37..3949d1af324 100644
--- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
+++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go
@@ -387,8 +387,11 @@ func (em *ElementManager) Bundles(ctx context.Context, 
upstreamCancelFn context.
                em.pendingElements.Wait()
                slog.Debug("no more pending elements: terminating pipeline")
                cancelFn(fmt.Errorf("elementManager out of elements, cleaning 
up"))
-               // Ensure the watermark evaluation goroutine exits.
+               // Ensure the watermark evaluation goroutine exits by locking 
the mutex
+               // before broadcasting, preventing a lost wake-up signal.
+               em.refreshCond.L.Lock()
                em.refreshCond.Broadcast()
+               em.refreshCond.L.Unlock()
        }()
        // Watermark evaluation goroutine.
        go func() {
@@ -2496,7 +2499,9 @@ func (em *ElementManager) wakeUpAt(t mtime.Time) {
                // only create this goroutine if we have real-time clock 
enabled (also implying the pipeline does not have TestStream).
                go func(fireAt time.Time) {
                        time.AfterFunc(time.Until(fireAt), func() {
+                               em.refreshCond.L.Lock()
                                em.refreshCond.Broadcast()
+                               em.refreshCond.L.Unlock()
                        })
                }(t.ToTime())
        }