[Failing Test]: TriggerAlways integration test failed in prism

[shunping]

What happened?

Trying to enable TriggerAlways primitive integration test (

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

Line 52 in 6077034

{pipeline: primitives.TriggerAlways},

), but it is not successful.

I am able to reproduce it with a standalone code

package main

import (
	"context"
	"log"
	"time"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window/trigger"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/testing/teststream"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/stats"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
)

func init() {}

func main() {
	beam.Init()

	ctx := context.Background()
	pipeline, s := beam.NewPipelineWithRoot()

	con := teststream.NewConfig()
	con.AddElements(1000, 1.0, 2.0, 3.0)
	con.AdvanceWatermark(11000)
	col := teststream.Create(s, con)
	windowSize := 10 * time.Second
	wfn := window.NewFixedWindows(windowSize)
	windowed := beam.WindowInto(s.Scope("Fixed"), wfn, col, beam.Trigger(trigger.Always()))

	sums := stats.Sum(s, windowed)
	debug.Print(s, sums)

	if err := beamx.Run(ctx, pipeline); err != nil {
		log.Fatalf("Failed to execute job: %v", err)
	}
}

The stats.Sum should sum on three window panes separately, since trigger.Always() is used. However, the result returns 6.

2025/09/14 15:28:17 INFO log from SDK worker worker.ID=job-001[go-job-1-1757878097559282000]_go worker.endpoint=localhost:53202 sdk.transformID=e7 sdk.location=.../sdks/go/pkg/beam/x/debug/print.go:54 sdk.time=2025-09-14T19:28:17.572Z sdk.msg="Elm: 6"

I changed stats.Sum to GroupByKey, and I can three three panes.

package main

import (
	"context"
	"log"
	"time"

	"github.com/apache/beam/sdks/v2/go/pkg/beam"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/core/graph/window/trigger"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/testing/teststream"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
	"github.com/apache/beam/sdks/v2/go/pkg/beam/x/debug"
)

func init() {}

func main() {
	beam.Init()

	ctx := context.Background()
	pipeline, s := beam.NewPipelineWithRoot()

	con := teststream.NewConfig()
	con.AddElements(1000, 1.0, 2.0, 3.0)
	con.AdvanceWatermark(11000)
	col := teststream.Create(s, con)
	windowSize := 10 * time.Second
	wfn := window.NewFixedWindows(windowSize)
	windowed := beam.WindowInto(s.Scope("Fixed"), wfn, col, beam.Trigger(trigger.Always()))

	kv := beam.ParDo(s, func(element float64) (string, float64) {
		return "0", element
	}, windowed)

	gbk := beam.GroupByKey(s, kv)
	debug.Print(s, gbk)

	if err := beamx.Run(ctx, pipeline); err != nil {
		log.Fatalf("Failed to execute job: %v", err)
	}
}

2025/09/14 15:32:38 INFO log from SDK worker worker.ID=job-001[go-job-1-1757878358434257000]_go worker.endpoint=localhost:53506 sdk.transformID=e5 sdk.location=.../sdks/go/pkg/beam/x/debug/print.go:77 sdk.time=2025-09-14T19:32:38.500Z sdk.msg="Elm: (0,[2])"
2025/09/14 15:32:38 INFO log from SDK worker worker.ID=job-001[go-job-1-1757878358434257000]_go worker.endpoint=localhost:53506 sdk.transformID=e5 sdk.location=.../sdks/go/pkg/beam/x/debug/print.go:77 sdk.time=2025-09-14T19:32:38.500Z sdk.msg="Elm: (0,[1])"
2025/09/14 15:32:38 INFO log from SDK worker worker.ID=job-001[go-job-1-1757878358434257000]_go worker.endpoint=localhost:53506 sdk.transformID=e5 sdk.location=.../sdks/go/pkg/beam/x/debug/print.go:77 sdk.time=2025-09-14T19:32:38.500Z sdk.msg="Elm: (0,[3])"

Issue Failure

Failure: Test is flaky

Issue Priority

Priority: 1 (unhealthy code / failing or flaky postcommit so we cannot be sure the product is healthy)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[shunping]

cc'ed @lostluck

[shunping]

The sum/combine behavior is the same when running a similar python pipeline.

import logging

logging.basicConfig(level=logging.INFO)

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam import window
from apache_beam.transforms.trigger import AfterWatermark, AfterProcessingTime, AccumulationMode, AfterCount
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms import trigger
import apache_beam as beam
from apache_beam import window
from apache_beam.transforms.trigger import AfterWatermark, AfterProcessingTime, AccumulationMode, AfterCount
from apache_beam.transforms.window import FixedWindows
from apache_beam.transforms import trigger
from apache_beam.transforms.periodicsequence import PeriodicImpulse
from apache_beam.transforms.periodicsequence import RebaseMode
from apache_beam.utils.timestamp import Timestamp
from apache_beam.testing.test_stream import TestStream


# prism runner option
options = PipelineOptions([
    "--streaming",
    "--environment_type=LOOPBACK",
    "--runner=PrismRunner",
    #"--runner=PortableRunner", "--job_endpoint=localhost:8073",
    #"--environment_type=LOOPBACK",
    "--experiments=prism_enable_rtc",
    #"--runner=PortableRunner", "--job_endpoint=localhost:8073",
    "--prism_log_level=info",
    # Unsafe trigger: `AfterProcessingTime(delay=1)` may lose data. Reason: MAY_FINISH. This can be overriden with the --allow_unsafe_triggers flag.
    "--allow_unsafe_triggers",
])


with beam.Pipeline(options=options) as p:
  _ = (
      p | TestStream().add_elements(elements=[1, 2, 3], event_timestamp=1000)
                      # .advance_watermark_to(2000)
                      # .add_elements(elements=[4, 5], event_timestamp=6000)
                      .advance_watermark_to(10000)
                      # .add_elements(elements=[10,], event_timestamp=52000)
                      # .advance_watermark_to(53000)
      | 'window' >> beam.WindowInto(
          FixedWindows(10 * 1000),
          #trigger=trigger.Repeatedly(trigger.AfterCount(3)),
          #trigger=trigger.AfterCount(2),
          #trigger=trigger.AfterWatermark(early=trigger.AfterCount(4)),
          #trigger=trigger.AfterWatermark(),
          trigger=trigger.Always(),
          accumulation_mode=trigger.AccumulationMode.DISCARDING)
      | beam.CombineGlobally(sum).without_defaults()
      | beam.LogElements(level=logging.WARNING,
          with_timestamp=True, with_window=True, with_pane_info=True, use_epoch_time=True))

This also yields 6 in the result.

[shunping]

Looks like the problem is caused by combiner lifting.

If I disable the combiner lifting here, then we will see three numbers (1, 2, 3) instead one (6).

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

Line 120 in 6077034

Combine(CombineCharacteristic{EnableLifting: true}),

[lostluck]

That sounds familiar.

Basically certain trigger combinations have to disable combiner lifting for correctness.