Fix the python test instead.

apache · Dec 4, 2024 · 7516b98 · 7516b98
1 parent 2e799bf
commit 7516b98
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 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
@@ -1579,7 +1579,7 @@ func (ss *stageState) updateWatermarks(em *ElementManager) set[string] {
 				// TODO(#https://github.com/apache/beam/issues/31438):
 				// Adjust with AllowedLateness
 				// Clear out anything we've already used.
-				if win.MaxTimestamp()+1 < newOut {
+				if win.MaxTimestamp() < newOut {
 					delete(wins, win)
 				}
 			}
@@ -1588,7 +1588,7 @@ func (ss *stageState) updateWatermarks(em *ElementManager) set[string] {
 			for win := range wins {
 				// TODO(#https://github.com/apache/beam/issues/31438):
 				// Adjust with AllowedLateness
-				if win.MaxTimestamp()+1 < newOut {
+				if win.MaxTimestamp() < newOut {
 					delete(wins, win)
 				}
 			}

diff --git a/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner_test.py b/sdks/python/apache_beam/runners/portability/fn_api_runner/fn_runner_test.py
@@ -776,7 +776,8 @@ def process(
           state.clear()
           yield buffer
         else:
-          timer.set(ts + 1)
+          # Set the timer to fire within it's window.
+          timer.set(ts + (1 - timestamp.Duration(micros=1000)))
 
       @userstate.on_timer(timer_spec)
       def process_timer(self, state=beam.DoFn.StateParam(state_spec)):