Construct trigger struct based on pipeline proto.

Author: shunping

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

@@ -573,4 +573,26 @@ 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 {
+	Delay         time.Duration
+	AlignToPeriod time.Duration
+	AlignToOffset time.Duration
+}
+
+func (t *TriggerAfterProcessingTime) onElement(input triggerInput, state *StateData) {}
+
+func (t *TriggerAfterProcessingTime) shouldFire(state *StateData) bool {
+	return false
+}
+
+func (t *TriggerAfterProcessingTime) onFire(state *StateData) {}
+
+func (t *TriggerAfterProcessingTime) reset(state *StateData) {}
+
+func (t *TriggerAfterProcessingTime) String() string {
+	return fmt.Sprintf("AfterProcessingTime[Delay: %v, AlignToPeriod: %v, AlignToOffset: %v]", t.Delay, t.AlignToPeriod, t.AlignToOffset)
+}
+
 // TODO https://github.com/apache/beam/issues/31438 Handle TriggerAfterProcessingTime

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

@@ -457,7 +457,25 @@ func buildTrigger(tpb *pipepb.Trigger) engine.Trigger {
 		}
 	case *pipepb.Trigger_Repeat_:
 		return &engine.TriggerRepeatedly{Repeated: buildTrigger(at.Repeat.GetSubtrigger())}
-	case *pipepb.Trigger_AfterProcessingTime_, *pipepb.Trigger_AfterSynchronizedProcessingTime_:
+	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]
+			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
+			}
+		}
+		return &engine.TriggerAfterProcessingTime{
+			Delay:         delay,
+			AlignToPeriod: period,
+			AlignToOffset: offset,
+		}
+	case *pipepb.Trigger_AfterSynchronizedProcessingTime_:
 		panic(fmt.Sprintf("unsupported trigger: %v", prototext.Format(tpb)))
 	default:
 		return &engine.TriggerDefault{}

sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go

@@ -316,7 +316,7 @@ func (s *Server) Prepare(ctx context.Context, req *jobpb.PrepareJobRequest) (_ *
 func hasUnsupportedTriggers(tpb *pipepb.Trigger) bool {
 	unsupported := false
 	switch at := tpb.GetTrigger().(type) {
-	case *pipepb.Trigger_AfterProcessingTime_, *pipepb.Trigger_AfterSynchronizedProcessingTime_:
+	case *pipepb.Trigger_AfterSynchronizedProcessingTime_:
 		return true
 	case *pipepb.Trigger_AfterAll_:
 		for _, st := range at.AfterAll.GetSubtriggers() {