From edbb7e0aa369ff41dd4e78d463083f803be34a56 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 3 Oct 2025 10:02:12 -0400 Subject: [PATCH 1/3] Support AfterSynchronizedProcessingTime trigger in prism. --- .../runners/prism/internal/engine/strategy.go | 30 +++++++++++++++++++ .../beam/runners/prism/internal/execute.go | 3 +- .../prism/internal/jobservices/management.go | 11 +++++-- .../prism/internal/unimplemented_test.go | 5 +--- 4 files changed, 40 insertions(+), 9 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go index 2aef5fcf332f..2eede05a3ec6 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go @@ -727,3 +727,33 @@ func (t *TriggerAfterProcessingTime) reset(state *StateData) { 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" +} diff --git a/sdks/go/pkg/beam/runners/prism/internal/execute.go b/sdks/go/pkg/beam/runners/prism/internal/execute.go index 9d23a89d4583..446472787520 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/execute.go +++ b/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" ) @@ -498,7 +497,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{} } diff --git a/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go b/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go index 12c3c42c2e92..54169b56bb51 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go +++ b/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go @@ -316,8 +316,13 @@ 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_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 +347,7 @@ func hasUnsupportedTriggers(tpb *pipepb.Trigger) bool { case *pipepb.Trigger_Repeat_: return hasUnsupportedTriggers(at.Repeat.GetSubtrigger()) default: - return false + return true } } diff --git a/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go b/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go index d54955f43d46..89cbd2b17f6c 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go +++ b/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 { From 3cba7b17231b819da26874c4271b768e1c8ddaf8 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 3 Oct 2025 11:29:25 -0400 Subject: [PATCH 2/3] Some minor bug fix in processing-time triggers. --- .../go/pkg/beam/runners/prism/internal/engine/strategy.go | 8 +++++--- sdks/go/pkg/beam/runners/prism/internal/execute.go | 1 + .../beam/runners/prism/internal/jobservices/management.go | 4 ++++ 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go index 2eede05a3ec6..691f249a5bea 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go @@ -718,9 +718,11 @@ 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) } diff --git a/sdks/go/pkg/beam/runners/prism/internal/execute.go b/sdks/go/pkg/beam/runners/prism/internal/execute.go index 446472787520..cad1fb7e5479 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/execute.go +++ b/sdks/go/pkg/beam/runners/prism/internal/execute.go @@ -387,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 } diff --git a/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go b/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go index 54169b56bb51..e3f65078657e 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go +++ b/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go @@ -314,6 +314,10 @@ 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) { // stateless leaf trigger From e08f207709730dd9d5a8c1b230a6d1f2119ca438 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Fri, 3 Oct 2025 11:32:50 -0400 Subject: [PATCH 3/3] Enable UsesTestStreamWithProcessingTime test category (20+ tests) and exclude 3 failed tests. --- runners/prism/java/build.gradle | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/runners/prism/java/build.gradle b/runners/prism/java/build.gradle index fd3631fd4a70..dbd7cc6cb5c0 100644 --- a/runners/prism/java/build.gradle +++ b/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' }