Address the reviewer feedback.

Author: shunping

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),