Remove advance watermark steps in tests.

Author: shunping

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

@@ -398,7 +398,7 @@ func (em *ElementManager) Bundles(ctx context.Context, upstreamCancelFn context.
 
 			// If there are no changed stages, ready processing time events,
 			// or injected bundles available, we wait until there are.
-			for len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles)+(len(em.testStreamHandler.events)-em.testStreamHandler.nextEventIndex) == 0 {
+			for len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles) == 0 {
 				// Check to see if we must exit
 				select {
 				case <-ctx.Done():
@@ -1633,7 +1633,8 @@ func (ss *stageState) startTriggeredBundle(em *ElementManager, key string, win t
 	)
 	slog.Debug("started a triggered bundle", "stageID", ss.ID, "bundleID", rb.BundleID, "size", len(toProcess))
 
-	//ss.bundlesToInject = append(ss.bundlesToInject, rb)
+	// TODO: Use ss.bundlesToInject rather than em.injectedBundles
+	// ss.bundlesToInject = append(ss.bundlesToInject, rb)
 	// Bundle is marked in progress here to prevent a race condition.
 	em.refreshCond.L.Lock()
 	em.injectedBundles = append(em.injectedBundles, rb)

sdks/go/test/integration/primitives/windowinto.go

@@ -217,42 +217,40 @@ func TriggerElementCount(s beam.Scope) {
 		}, 2)
 }
 
-// TriggerAfterProcessingTime tests the AfterProcessingTime Trigger, it fires output panes once 't' processing time has passed
+// TriggerAfterProcessingTimeNotTriggered tests the AfterProcessingTime Trigger. It won't fire because 't' processing time is not reached
 // Not yet supported by the flink runner:
 // java.lang.UnsupportedOperationException: Advancing Processing time is not supported by the Flink Runner.
-func TriggerAfterProcessingTime(s beam.Scope) {
+func TriggerAfterProcessingTimeNotTriggered(s beam.Scope) {
 	con := teststream.NewConfig()
 	con.AdvanceProcessingTime(100)
 	con.AddElements(1000, 1.0, 2.0, 3.0)
-	con.AdvanceWatermark(1000)      // must advance watermark so that elements are processed in the downstream stages
-	con.AdvanceProcessingTime(5000) // advance processing time to fire the trigger
-	con.AddElements(2000, 4.0)
-	con.AdvanceWatermark(2000)
+	con.AdvanceProcessingTime(4999) // advance processing time but not enough to fire the trigger
+	con.AddElements(22000, 4.0)
+
 	col := teststream.Create(s, con)
 
 	validateEquals(s.Scope("Global"), window.NewGlobalWindows(), col,
 		[]beam.WindowIntoOption{
 			beam.Trigger(trigger.AfterProcessingTime().PlusDelay(5 * time.Second)),
-		}, 6.0, 4.0)
+		}, 10.0)
 }
 
-// TriggerAfterProcessingTime tests the AfterProcessingTime Trigger, it fires output panes once 't' processing time has passed
+// TriggerAfterProcessingTime tests the AfterProcessingTime Trigger. It fires output panes once 't' processing time has passed
 // Not yet supported by the flink runner:
 // java.lang.UnsupportedOperationException: Advancing Processing time is not supported by the Flink Runner.
-func TriggerAfterProcessingTimeNotTriggered(s beam.Scope) {
+func TriggerAfterProcessingTime(s beam.Scope) {
 	con := teststream.NewConfig()
 	con.AdvanceProcessingTime(100)
 	con.AddElements(1000, 1.0, 2.0, 3.0)
-	con.AdvanceWatermark(1000)      // must advance watermark so that elements are processed in the downstream stages
-	con.AdvanceProcessingTime(4999) // advance processing time but not enough to fire the trigger
-	con.AddElements(2000, 4.0)
-	con.AdvanceWatermark(2000)
+	con.AdvanceProcessingTime(5000) // advance processing time to fire the trigger
+	con.AddElements(22000, 4.0)
+
 	col := teststream.Create(s, con)
 
 	validateEquals(s.Scope("Global"), window.NewGlobalWindows(), col,
 		[]beam.WindowIntoOption{
 			beam.Trigger(trigger.AfterProcessingTime().PlusDelay(5 * time.Second)),
-		}, 10.0)
+		}, 6.0, 4.0)
 }
 
 // TriggerRepeat tests the repeat trigger. As of now is it is configure to take only one trigger as a subtrigger.