Handle the case when after-processing-time trigger is called repeated.

Author: shunping

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

@@ -589,8 +589,9 @@ type TriggerAfterProcessingTime struct {
 }
 
 type afterProcessingTimeState struct {
-	emNow      mtime.Time
-	firingTime mtime.Time
+	emNow              mtime.Time
+	firingTime         mtime.Time
+	endOfWindowReached bool
 }
 
 func (t *TriggerAfterProcessingTime) onElement(input triggerInput, state *StateData) {
@@ -601,12 +602,14 @@ func (t *TriggerAfterProcessingTime) onElement(input triggerInput, state *StateD
 
 	if ts.extra == nil {
 		ts.extra = afterProcessingTimeState{
-			emNow:      input.emNow,
-			firingTime: t.applyTimestampTransforms(input.emNow),
+			emNow:              input.emNow,
+			firingTime:         t.applyTimestampTransforms(input.emNow),
+			endOfWindowReached: input.endOfWindowReached,
 		}
 	} else {
 		s, _ := ts.extra.(afterProcessingTimeState)
 		s.emNow = input.emNow
+		s.endOfWindowReached = input.endOfWindowReached
 		ts.extra = s
 	}
 
@@ -634,7 +637,7 @@ func (t *TriggerAfterProcessingTime) applyTimestampTransforms(start mtime.Time)
 
 func (t *TriggerAfterProcessingTime) shouldFire(state *StateData) bool {
 	ts := state.getTriggerState(t)
-	if ts.extra == nil {
+	if ts.extra == nil || ts.finished {
 		return false
 	}
 	s := ts.extra.(afterProcessingTimeState)
@@ -646,11 +649,28 @@ func (t *TriggerAfterProcessingTime) onFire(state *StateData) {
 	if ts.finished {
 		return
 	}
-	triggerClearAndFinish(t, state)
+
+	// We don't reset the state here, only mark it as finished
+	ts.finished = true
+	state.setTriggerState(t, ts)
 }
 
 func (t *TriggerAfterProcessingTime) reset(state *StateData) {
-	delete(state.Trigger, t)
+	ts := state.getTriggerState(t)
+	if ts.extra != nil {
+		if ts.extra.(afterProcessingTimeState).endOfWindowReached {
+			delete(state.Trigger, t)
+			return
+		}
+	}
+
+	// Not reaching the end of window yet.
+	// We keep the state (especially the next possible firing time) in case the trigger is called again
+	ts.finished = false
+	s := ts.extra.(afterProcessingTimeState)
+	s.firingTime = t.applyTimestampTransforms(s.firingTime) // compute next possible firing time
+	ts.extra = s
+	state.setTriggerState(t, ts)
 }
 
 func (t *TriggerAfterProcessingTime) String() string {

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

@@ -488,11 +488,30 @@ func TestTriggers_isReady(t *testing.T) {
 				{triggerInput{emNow: 0}, false},
 				{triggerInput{emNow: 1000}, false},
 				{triggerInput{emNow: 2000}, false},
-				{triggerInput{emNow: 3000}, true},  // first the first time
-				{triggerInput{emNow: 4000}, false}, // trigger firing time is set again
+				{triggerInput{emNow: 3000}, true}, // firing the first time, trigger set again
+				{triggerInput{emNow: 4000}, false},
 				{triggerInput{emNow: 5000}, false},
-				{triggerInput{emNow: 6000}, false},
-				{triggerInput{emNow: 7000}, true}, // trigger firing again
+				{triggerInput{emNow: 6000}, true}, // firing the second time
+			},
+		}, {
+			name: "afterProcessingTime_Repeated_AcrossWindows", trig: &TriggerRepeatedly{
+				&TriggerAfterProcessingTime{
+					Transforms: []TimestampTransform{
+						{Delay: 3 * time.Second},
+					}}},
+			inputs: []io{
+				{triggerInput{emNow: 0}, false},
+				{triggerInput{emNow: 1000}, false},
+				{triggerInput{emNow: 2000}, false},
+				{triggerInput{emNow: 3000}, true}, // fire the first time, trigger is set again
+				{triggerInput{emNow: 4000}, false},
+				{triggerInput{emNow: 5000}, false},
+				{triggerInput{emNow: 6000,
+					endOfWindowReached: true}, true}, // fire the second time, reach end of window and start over
+				{triggerInput{emNow: 7000}, false}, // trigger firing time is set to 7s + 3s = 10s
+				{triggerInput{emNow: 8000}, false},
+				{triggerInput{emNow: 9000}, false},
+				{triggerInput{emNow: 10000}, true}, // fire in the new window
 			},
 		}, {
 			name: "default",