[Prism] Support AfterSynchronizedProcessingTime and enable java processing-time trigger tests

runners/prism/java/build.gradle

@@ -86,13 +86,9 @@ def sickbayTests = [
     'org.apache.beam.sdk.metrics.MetricsTest$CommittedMetricTests.testCommittedStringSetMetrics',
     'org.apache.beam.sdk.metrics.MetricsTest$CommittedMetricTests.testCommittedGaugeMetrics',
 
-    // ProcessingTime triggers not yet implemented in Prism.
-    // https://github.com/apache/beam/issues/31438
+    // negative WaitGroup counter when failing bundle
     'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testAfterProcessingTimeContinuationTriggerUsingState',
-    'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testCombiningAccumulatingProcessingTime',
-    'org.apache.beam.sdk.transforms.GroupByKeyTest$BasicTests.testAfterProcessingTimeContinuationTriggerEarly',
-    'org.apache.beam.sdk.testing.TestStreamTest.testProcessingTimeTrigger',
-    'org.apache.beam.sdk.testing.TestStreamTest.testLateDataAccumulating', // Uses processing time trigger for early firings.
+    'org.apache.beam.sdk.testing.TestStreamTest.testEarlyPanesOfWindow',
 
     // A regression introduced when we use number of pending elements rather than watermark to determine
     // the bundle readiness of a stateless stage.
@@ -107,6 +103,7 @@ def sickbayTests = [
     // Triggered Side Inputs not yet implemented in Prism.
     // https://github.com/apache/beam/issues/31438
     'org.apache.beam.sdk.transforms.ViewTest.testTriggeredLatestSingleton',
+    'org.apache.beam.sdk.testing.TestStreamTest.testProcessingTimeTrigger',
 
     // Prism doesn't support multiple TestStreams.
     'org.apache.beam.sdk.testing.TestStreamTest.testMultipleStreams',
@@ -116,6 +113,9 @@ def sickbayTests = [
     // GroupIntoBatchesTest tests that fail:
     // Teststream has bad KV encodings due to using an outer context.
     'org.apache.beam.sdk.transforms.GroupIntoBatchesTest.testInStreamingMode',
+    'org.apache.beam.sdk.transforms.GroupIntoBatchesTest.testBufferingTimerInFixedWindow',
+    // sdk worker disconnected
+    'org.apache.beam.sdk.transforms.GroupIntoBatchesTest.testBufferingTimerInGlobalWindow',
     // ShardedKey not yet implemented.
     'org.apache.beam.sdk.transforms.GroupIntoBatchesTest.testWithShardedKeyInGlobalWindow',
 
@@ -204,10 +204,6 @@ def createPrismValidatesRunnerTask = { name, environmentType ->
       // https://github.com/apache/beam/issues?q=is%3Aissue+is%3Aopen+MultimapState
       excludeCategories 'org.apache.beam.sdk.testing.UsesMultimapState'
 
-      // Processing time with TestStream is unreliable without being able to control
-      // SDK side time portably. Ignore these tests.
-      excludeCategories 'org.apache.beam.sdk.testing.UsesTestStreamWithProcessingTime'
-
       // Not yet supported in Prism.
       excludeCategories 'org.apache.beam.sdk.testing.UsesBoundedTrieMetrics'
     }

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

@@ -718,12 +718,44 @@ func (t *TriggerAfterProcessingTime) reset(state *StateData) {
 	// 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.emNow) // compute next possible firing time
-	ts.extra = s
+	if ts.extra != nil {
+		s := ts.extra.(afterProcessingTimeState)
+		s.firingTime = t.applyTimestampTransforms(s.emNow) // compute next possible firing time
+		ts.extra = s
+	}
 	state.setTriggerState(t, ts)
 }
 
 func (t *TriggerAfterProcessingTime) String() string {
 	return fmt.Sprintf("AfterProcessingTime[%v]", t.Transforms)
 }
+
+// TriggerAfterSynchronizedProcessingTime is supposed to fires once when processing
+// time across multiple workers synchronizes with the first element's processing time.
+// It is a no-op in the current prism single-node architecture, because we only have
+// one worker/machine. Therefore, the trigger just fires once it receives the data.
+type TriggerAfterSynchronizedProcessingTime struct{}
+
+func (t *TriggerAfterSynchronizedProcessingTime) onElement(triggerInput, *StateData) {}
+
+func (t *TriggerAfterSynchronizedProcessingTime) shouldFire(state *StateData) bool {
+	ts := state.getTriggerState(t)
+	return !ts.finished
+}
+
+func (t *TriggerAfterSynchronizedProcessingTime) onFire(state *StateData) {
+	if !t.shouldFire(state) {
+		return
+	}
+	ts := state.getTriggerState(t)
+	ts.finished = true
+	state.setTriggerState(t, ts)
+}
+
+func (t *TriggerAfterSynchronizedProcessingTime) reset(state *StateData) {
+	delete(state.Trigger, t)
+}
+
+func (t *TriggerAfterSynchronizedProcessingTime) String() string {
+	return "AfterSynchronizedProcessingTime"
+}

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

@@ -37,7 +37,6 @@ import (
 	"github.com/apache/beam/sdks/v2/go/pkg/beam/runners/prism/internal/worker"
 	"golang.org/x/exp/maps"
 	"golang.org/x/sync/errgroup"
-	"google.golang.org/protobuf/encoding/prototext"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -388,6 +387,7 @@ func executePipeline(ctx context.Context, wks map[string]*worker.W, j *jobservic
 				wk := wks[s.envID]
 				if err := s.Execute(ctx, j, wk, comps, em, rb); err != nil {
 					// Ensure we clean up on bundle failure
+					j.Logger.Error("Bundle Failed.", slog.Any("error", err))
 					em.FailBundle(rb)
 					return err
 				}
@@ -498,7 +498,7 @@ func buildTrigger(tpb *pipepb.Trigger) engine.Trigger {
 			Transforms: transforms,
 		}
 	case *pipepb.Trigger_AfterSynchronizedProcessingTime_:
-		panic(fmt.Sprintf("unsupported trigger: %v", prototext.Format(tpb)))
+		return &engine.TriggerAfterSynchronizedProcessingTime{}
 	default:
 		return &engine.TriggerDefault{}
 	}

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

@@ -314,10 +314,19 @@ func (s *Server) Prepare(ctx context.Context, req *jobpb.PrepareJobRequest) (_ *
 }
 
 func hasUnsupportedTriggers(tpb *pipepb.Trigger) bool {
+	if tpb == nil {
+		return false
+	}
+
 	unsupported := false
 	switch at := tpb.GetTrigger().(type) {
-	case *pipepb.Trigger_AfterSynchronizedProcessingTime_:
-		return true
+	// stateless leaf trigger
+	case *pipepb.Trigger_Never_, *pipepb.Trigger_Always_, *pipepb.Trigger_Default_:
+		return false
+	// stateful leaf trigger
+	case *pipepb.Trigger_ElementCount_, *pipepb.Trigger_AfterProcessingTime_, *pipepb.Trigger_AfterSynchronizedProcessingTime_:
+		return false
+	// composite trigger below
 	case *pipepb.Trigger_AfterAll_:
 		for _, st := range at.AfterAll.GetSubtriggers() {
 			unsupported = unsupported || hasUnsupportedTriggers(st)
@@ -342,7 +351,7 @@ func hasUnsupportedTriggers(tpb *pipepb.Trigger) bool {
 	case *pipepb.Trigger_Repeat_:
 		return hasUnsupportedTriggers(at.Repeat.GetSubtrigger())
 	default:
-		return false
+		return true
 	}
 }
 

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

@@ -49,10 +49,6 @@ func TestUnimplemented(t *testing.T) {
 		// See https://github.com/apache/beam/issues/31153.
 		{pipeline: primitives.TriggerElementCount},
 		{pipeline: primitives.TriggerOrFinally},
-
-		// Currently unimplemented triggers.
-		// https://github.com/apache/beam/issues/31438
-		{pipeline: primitives.TriggerAfterSynchronizedProcessingTime},
 	}
 
 	for _, test := range tests {
@@ -94,6 +90,7 @@ func TestImplemented(t *testing.T) {
 		{pipeline: primitives.TriggerRepeat},
 		{pipeline: primitives.TriggerAfterProcessingTime},
 		{pipeline: primitives.TriggerAfterProcessingTimeNotTriggered},
+		{pipeline: primitives.TriggerAfterSynchronizedProcessingTime},
 	}
 
 	for _, test := range tests {