Fill in callbacks for TriggerAfterProcessingTime

Author: shunping

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

@@ -573,26 +573,69 @@ func (t *TriggerDefault) String() string {
 	return "Default"
 }
 
-// TriggerAfterProcessingTime fires once after a specified amount of processing time
-// has passed since an element was first seen.
-type TriggerAfterProcessingTime struct {
+// TimestampTransform is the engine's representation of a processing time transform.
+type TimestampTransform struct {
 	Delay         time.Duration
 	AlignToPeriod time.Duration
 	AlignToOffset time.Duration
 }
 
-func (t *TriggerAfterProcessingTime) onElement(input triggerInput, state *StateData) {}
+// TriggerAfterProcessingTime fires once after a specified amount of processing time
+// has passed since an element was first seen.
+// Uses the extra state field to track if the time has been reached.
+type TriggerAfterProcessingTime struct {
+	Transforms []TimestampTransform
+}
+
+func (t *TriggerAfterProcessingTime) onElement(input triggerInput, state *StateData) {
+	ts := state.getTriggerState(t)
+	if ts.finished {
+		return
+	}
+
+	if ts.extra == nil {
+		ts.extra = mtime.Now()
+	}
+
+	state.setTriggerState(t, ts)
+}
+
+func (t *TriggerAfterProcessingTime) applyTimestampTransforms(start mtime.Time) mtime.Time {
+	ret := start
+	for _, transform := range t.Transforms {
+		ret = ret + mtime.Time(transform.Delay/time.Millisecond)
+		if transform.AlignToPeriod > 0 {
+			// Formula from https://cloud.google.com/blog/products/data-analytics/windowing-and-triggering-in-apache-beam
+			// timestamp - (timestamp % period) + period
+			// And with an offset, we adjust before and after.
+			tsMs := ret
+			periodMs := mtime.Time(transform.AlignToPeriod / time.Millisecond)
+			offsetMs := mtime.Time(transform.AlignToOffset / time.Millisecond)
+
+			adjustedMs := tsMs - offsetMs
+			alignedMs := adjustedMs - (adjustedMs % periodMs) + periodMs + offsetMs
+			ret = alignedMs
+		}
+	}
+	return ret
+}
 
 func (t *TriggerAfterProcessingTime) shouldFire(state *StateData) bool {
-	return false
+	ts := state.getTriggerState(t)
+	if ts.extra == nil {
+		return false
+	}
+	startTime := ts.extra.(mtime.Time)
+	firingTime := t.applyTimestampTransforms(startTime)
+	return mtime.Now() > firingTime
 }
 
 func (t *TriggerAfterProcessingTime) onFire(state *StateData) {}
 
-func (t *TriggerAfterProcessingTime) reset(state *StateData) {}
+func (t *TriggerAfterProcessingTime) reset(state *StateData) {
+	delete(state.Trigger, t)
+}
 
 func (t *TriggerAfterProcessingTime) String() string {
-	return fmt.Sprintf("AfterProcessingTime[Delay: %v, AlignToPeriod: %v, AlignToOffset: %v]", t.Delay, t.AlignToPeriod, t.AlignToOffset)
+	return fmt.Sprintf("AfterProcessingTime[%v]", t.Transforms)
 }
-
-// TODO https://github.com/apache/beam/issues/31438 Handle TriggerAfterProcessingTime

sdks/go/pkg/beam/runners/prism/internal/execute.go

@@ -458,22 +458,24 @@ func buildTrigger(tpb *pipepb.Trigger) engine.Trigger {
 	case *pipepb.Trigger_Repeat_:
 		return &engine.TriggerRepeatedly{Repeated: buildTrigger(at.Repeat.GetSubtrigger())}
 	case *pipepb.Trigger_AfterProcessingTime_:
-		var delay, period, offset time.Duration
-		// TODO: support multiple transforms.
-		if len(at.AfterProcessingTime.GetTimestampTransforms()) > 0 {
-			ts := at.AfterProcessingTime.GetTimestampTransforms()[0]
+		var transforms []engine.TimestampTransform
+		for _, ts := range at.AfterProcessingTime.GetTimestampTransforms() {
+			var delay, period, offset time.Duration
 			if d := ts.GetDelay(); d != nil {
 				delay = time.Duration(d.GetDelayMillis()) * time.Millisecond
 			}
 			if align := ts.GetAlignTo(); align != nil {
 				period = time.Duration(align.GetPeriod()) * time.Millisecond
 				offset = time.Duration(align.GetOffset()) * time.Millisecond
 			}
+			transforms = append(transforms, engine.TimestampTransform{
+				Delay:         delay,
+				AlignToPeriod: period,
+				AlignToOffset: offset,
+			})
 		}
 		return &engine.TriggerAfterProcessingTime{
-			Delay:         delay,
-			AlignToPeriod: period,
-			AlignToOffset: offset,
+			Transforms: transforms,
 		}
 	case *pipepb.Trigger_AfterSynchronizedProcessingTime_:
 		panic(fmt.Sprintf("unsupported trigger: %v", prototext.Format(tpb)))