From 471708b8e0495106b8a5d7931774a7fbe4a61f5f Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 12 Sep 2025 10:08:00 -0400 Subject: [PATCH 1/2] Add preRefreshedInput watermark and use it in bundleReady. --- .../runners/prism/internal/engine/elementmanager.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) 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 8c8b71ca4146..9e3a2ce80917 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -1153,6 +1153,7 @@ type stageState struct { input mtime.Time // input watermark for the parallel input. output mtime.Time // Output watermark for the whole stage estimatedOutput mtime.Time // Estimated watermark output from DoFns + preRefreshedInput mtime.Time // input watermark before the last watermark refresh pending elementHeap // pending input elements for this stage that are to be processesd inprogress map[string]elements // inprogress elements by active bundles, keyed by bundle @@ -2014,6 +2015,8 @@ func (ss *stageState) updateWatermarks(em *ElementManager) set[string] { newIn = minPending } + ss.preRefreshedInput = ss.input + // If bigger, advance the input watermark. if newIn > ss.input { ss.input = newIn @@ -2171,12 +2174,13 @@ func (ss *stageState) bundleReady(em *ElementManager, emNow mtime.Time) (mtime.T ptimeEventsReady := ss.processingTimeTimers.Peek() <= emNow || emNow == mtime.MaxTimestamp injectedReady := len(ss.bundlesToInject) > 0 - // If the upstream watermark and the input watermark are the same, + // If the upstream watermark, the input watermark, and the input watermark before the last refresh are the same, // then we can't yet process this stage. inputW := ss.input _, upstreamW := ss.UpstreamWatermark() - if inputW == upstreamW { - slog.Debug("bundleReady: unchanged upstream watermark", + preRefreshInputW := ss.preRefreshedInput + if inputW == upstreamW && preRefreshInputW == inputW { + slog.Debug("bundleReady: unchanged upstream/input watermark", slog.String("stage", ss.ID), slog.Group("watermark", slog.Any("upstream", upstreamW), From 17cad7d51f041c2ddf96fdffcbc797558d1ff349 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 12 Sep 2025 15:02:06 -0400 Subject: [PATCH 2/2] Address the reviewer feedback. --- .../prism/internal/engine/elementmanager.go | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) 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 9e3a2ce80917..18f10f45e6ca 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -1153,7 +1153,7 @@ type stageState struct { input mtime.Time // input watermark for the parallel input. output mtime.Time // Output watermark for the whole stage estimatedOutput mtime.Time // Estimated watermark output from DoFns - preRefreshedInput mtime.Time // input watermark before the last watermark refresh + previousInput mtime.Time // input watermark before the latest watermark refresh pending elementHeap // pending input elements for this stage that are to be processesd inprogress map[string]elements // inprogress elements by active bundles, keyed by bundle @@ -2015,7 +2015,7 @@ func (ss *stageState) updateWatermarks(em *ElementManager) set[string] { newIn = minPending } - ss.preRefreshedInput = ss.input + ss.previousInput = ss.input // If bigger, advance the input watermark. if newIn > ss.input { @@ -2174,13 +2174,14 @@ func (ss *stageState) bundleReady(em *ElementManager, emNow mtime.Time) (mtime.T ptimeEventsReady := ss.processingTimeTimers.Peek() <= emNow || emNow == mtime.MaxTimestamp injectedReady := len(ss.bundlesToInject) > 0 - // If the upstream watermark, the input watermark, and the input watermark before the last refresh are the same, - // then we can't yet process this stage. + // If the upstream watermark does not change, we can't yet process this stage. + // To check whether upstream water is unchanged, we evaluate if the input watermark, and + // the input watermark before the latest refresh are the same. inputW := ss.input _, upstreamW := ss.UpstreamWatermark() - preRefreshInputW := ss.preRefreshedInput - if inputW == upstreamW && preRefreshInputW == inputW { - slog.Debug("bundleReady: unchanged upstream/input watermark", + previousInputW := ss.previousInput + if inputW == upstreamW && previousInputW == inputW { + slog.Debug("bundleReady: unchanged upstream watermark", slog.String("stage", ss.ID), slog.Group("watermark", slog.Any("upstream", upstreamW),