persicoana opened a new issue, #40818: URL: https://github.com/apache/beam/issues/40818
### What happened? Prism admits expired aggregation input while pending work in a later window holds its output watermark behind upstream progress. The job reaches `DONE`, but its physical output includes data for a window already past allowed lateness. Reproduced on current master `49a6a2ecfa4679c025e434deb90293dc9cfee042` (2.78.0 development), built with Go 1.26.2 on Linux x86_64, using Apache Beam Python 2.77.0 / CPython 3.12. The same reduction also reproduces on the 2.77.0 source. No SQL, external source, custom provider, or wall-clock simulation is required. The finite TestStream has four events: 1. B=4 at timestamp 11 seconds. 2. Advance the watermark to 16 seconds. 3. A=100 at timestamp 5 seconds. 4. Advance to infinity. Apply fixed 10-second windows, accumulating AfterWatermark with late AfterCount(1), allowed lateness 5 seconds, and end-of-window timestamps, then GroupByKey. Expected: A in [0,10) is expired and absent; B in [10,20) remains. Actual unmodified master: A=100 is emitted twice, alongside B's on-time and closing panes. A local correction emits only B's two panes. The reproducer below writes every native pane and fails the original result even though the terminal state is DONE. The aggregate admission threshold in `internal/engine/elementmanager.go` is currently `ss.output`. Pending later-window work holds that behind the input PCollection watermark. I plan to propose using the existing minimum-upstream-watermark accessor, with five engine controls for held-output expiration, within-lateness, exact completion, past completion and the minimum of two upstream inputs. This keeps internal/output holds and the separate stateful path unchanged. Related context: #31438, #33515 and #33542. Searches for expired/pending/late input and upstream-watermark corrections did not identify this minimal case; please redirect if it is already tracked. ### Reproduction Build unmodified master Prism from `sdks` using `go build -o /path/to/prism ./go/cmd/prism`. With `apache-beam==2.77.0` installed, save the following as `reproduce.py` and run: ```sh python reproduce.py --prism /path/to/prism --directory /new/scratch/output ``` The output directory must not exist. Expected original exit is nonzero with a semantic failure; each physical pane and the runner/source hashes are retained in `report.json`. A finite-DONE result is not used as a complete continuous-stream claim. <details><summary>Self-contained SDK reproducer</summary> ```python # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Minimal SDK-only expired-late grouping control, no SQL/application imports.""" import argparse import hashlib import json from pathlib import Path import sys import threading import uuid def reject_app(event, args): if event == 'import' and args[0].split('.')[0] in ['beedrill', 'scripts']: raise ImportError('SDK-only reduction') sys.addaudithook(reject_app) import apache_beam as beam from apache_beam.coders import TupleCoder, StrUtf8Coder, VarIntCoder from apache_beam.options.pipeline_options import PipelineOptions from apache_beam.testing.test_stream import TestStream from apache_beam.transforms import window, trigger from google.protobuf.json_format import MessageToDict class Observe(beam.DoFn): def __init__(self, directory): self.directory = directory def process( self, kv, stamp=beam.DoFn.TimestampParam, win=beam.DoFn.WindowParam, pane=beam.DoFn.PaneInfoParam, ): values = list(kv[1]) record = dict( key=kv[0], values=sorted(values), count=len(values), total=sum(values), timestamp_us=stamp.micros, start_us=win.start.micros, end_us=win.end.micros, pane=dict( index=pane.index, first=pane.is_first, last=pane.is_last, timing=int(pane.timing), nonspeculative_index=pane.nonspeculative_index, ), ) p = Path(self.directory) p.mkdir(parents=True, exist_ok=True) (p / (uuid.uuid4().hex + '.json')).write_text(json.dumps(record, sort_keys=True) + '\n') return [] def main(): a = argparse.ArgumentParser() a.add_argument('--directory', type=Path, required=True) a.add_argument('--prism', type=Path, required=True) a.add_argument('--runner', choices=['PrismRunner', 'DirectRunner'], default='PrismRunner') args = a.parse_args() args.directory.mkdir(parents=True, exist_ok=False) report = dict( binary_sha256=hashlib.sha256(args.prism.read_bytes()).hexdigest(), runner=args.runner, beam=beam.__version__, source_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), application_imported=False, stage='construction', ) try: options = ['--runner=' + args.runner, '--streaming'] if args.runner == 'PrismRunner': options += [ '--prism_location=' + str(args.prism.resolve()), '--environment_type=LOOPBACK', ] p = beam.Pipeline(options=PipelineOptions(options)) coder = TupleCoder([StrUtf8Coder(), VarIntCoder()]) ts = ( TestStream(coder=coder) .add_elements([window.TimestampedValue(('B', 4), 11)]) .advance_watermark_to(16) .add_elements([window.TimestampedValue(('A', 100), 5)]) .advance_watermark_to_infinity() ) values = ( p | ts.with_output_types(beam.typehints.KV[str, int]) | beam.WindowInto( window.FixedWindows(10), trigger=trigger.AfterWatermark(late=trigger.AfterCount(1)), accumulation_mode=trigger.AccumulationMode.ACCUMULATING, allowed_lateness=5, timestamp_combiner=window.TimestampCombiner.OUTPUT_AT_EOW, ) ) grouped = values | beam.GroupByKey() proto, ctx = p.to_runner_api(return_context=True) pc = proto.components.pcollections[ctx.pcollections.get_id(grouped)] report['strategy'] = MessageToDict( proto.components.windowing_strategies[pc.windowing_strategy_id] ) report['is_bounded'] = grouped.is_bounded grouped | beam.ParDo(Observe(str((args.directory / 'physical').resolve()))) (args.directory / 'pipeline.pb').write_bytes( p.to_runner_api().SerializeToString(deterministic=True) ) report['stage'] = 'execution' job = p.run() def timeout(): report['watchdog_fired'] = True try: report['cancel_state'] = str(job.cancel()) except Exception as exc: report['cancel_error'] = str(exc) timer = threading.Timer(120, timeout) timer.daemon = True timer.start() try: report['state'] = str(job.wait_until_finish()) finally: timer.cancel() report['physical'] = [ json.loads(f.read_text()) for f in sorted((args.directory / 'physical').glob('*.json')) ] if report['state'] != 'DONE' or report.get('watchdog_fired'): raise RuntimeError('Runner did not complete') rows = report['physical'] if not rows or any( row['key'] != 'B' or row['values'] != [4] or row['start_us'] != 10000000 or row['end_us'] != 20000000 for row in rows ): raise RuntimeError('Semantic failure: expired A must be absent and B retained') report['stage'] = 'complete' except BaseException as exc: report.update(error_type=type(exc).__name__, error=str(exc)) raise finally: (args.directory / 'report.json').write_text( json.dumps(report, indent=2, sort_keys=True) + '\n' ) if __name__ == '__main__': main() ``` </details> ### Issue Priority Priority: 2 (default / most bugs should be filed as P2) ### Issue Components - [x] 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 - [x] Component: Prism Runner - [ ] Component: Twister2 Runner - [ ] Component: Hazelcast Jet Runner - [ ] Component: Google Cloud Dataflow Runner -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
