[Failing Test]: Data race detected on TriggerAfterProcessingTimeNotTriggered

[shunping]

What happened?

https://github.com/apache/beam/actions/runs/18208074497/job/51842943177

WARNING: DATA RACE
Write at 0x00c00081c928 by goroutine 161:
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*mtimeHeap).Push()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/holds.go:38 +0xcc
  container/heap.Push()
      /Users/shunping/sdk/go1.24.4/src/container/heap/heap.go:52 +0x48
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*stageRefreshQueue).Schedule()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/processingtime.go:71 +0x190
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*stageState).injectTriggeredBundlesIfReady()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:1369 +0x960
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*aggregateStageKind).addPending()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:1425 +0x500
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*stageState).AddPending()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:1307 +0xc0
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*ElementManager).PersistBundle()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:877 +0xa04
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal.(*stage).Execute()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/stage.go:350 +0x2924
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal.executePipeline.func4()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/execute.go:389 +0x114
  golang.org/x/sync/errgroup.(*Group).Go.func1()
      ../go/pkg/mod/golang.org/x/sync@v0.17.0/errgroup/errgroup.go:93 +0x7c

Previous read at 0x00c00081c928 by goroutine 88:
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*stageRefreshQueue).AdvanceTo()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/processingtime.go:88 +0x68
  github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*ElementManager).Bundles.func2()
      ..beam-dev-go-prism/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:396 +0x15c

Notice that stageRefreshQueue is not goroutine safe.

beam/sdks/go/pkg/beam/runners/prism/internal/engine/processingtime.go

Line 46 in 3534960

// stageRefreshQueue is not goroutine safe and relies on external synchronization.

The race condition is due to the fact that em.processingTimeEvents can be accessed in the watermark evaluation goroutine and the goroutine that executes a stage (where the PersistBundle and addPending are called).

Issue Failure

Failure: Test is flaky

Issue Priority

Priority: 1 (unhealthy code / failing or flaky postcommit so we cannot be sure the product is healthy)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[shunping]

After fixing the race condition, I see another memory issue

"invalid memory address or nil pointer dereference" 
traceback:=
goroutine 24081 [running]:
runtime/debug.Stack()
  ../sdk/go1.24.4/src/runtime/debug/stack.go:26 +0x6c
github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*ElementManager).Bundles.func2.1()
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:388 +0x64\npanic({0x1047df560?, 0x109116e10?})
  ../sdk/go1.24.4/src/runtime/panic.go:792 +0x124
github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*stageState).buildTriggeredBundle(0xc001969860, 0xc001e84598?, {0xc000ff3330, 0x2}, {0x105778998, 0x1091ac0c0})
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:1584 +0xcc
github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*aggregateStageKind).buildProcessingTimeBundle.func1({{0x105778998, 0x1091ac0c0}, 0xffffc77cedd33bec, 0x20c49ba0bcf7f7, {0x3, 0x1, 0x1, 0x0, 0x0}, {0xc0019c2947, ...}, ...}, ...)
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:2100 +0x2f4
github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.handleProcessingTimeTimer(0xc001969860, 0xc001dec0c0, 0xffffc77cedd33bec, 0xc000af5a78)
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:2049 +0xa58
github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*aggregateStageKind).buildProcessingTimeBundle(0xc001969920?, 0xc001969860, 0xc001dec0c0, 0xffffc77cedd33bec)
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:2085 +0x8c
github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*stageState).startProcessingTimeBundle(0xc001969860, 0xc001dec0c0, 0xffffc77cedd33bec, 0xc00169fb30)
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:1992 +0xc4
github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*ElementManager).Bundles.func2()
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:485 +0x9d8
created by github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/engine.(*ElementManager).Bundles in goroutine 24071
  ../sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:383 +0x29c