francisduffy opened a new issue, #26190: URL: https://github.com/apache/beam/issues/26190
### What needs to happen? The minimal working example below works (with `DirectRunner`) with Python 3.9 and Apache Beam 2.38.0 but fails on Apache Beam 2.39.0 and 2.44.0. Using Apache Beam 2.39.0 and 2.44.0, the example fails with the error `AssertionError: A total of 2 watermark-pending bundles did not execute.`. When I switch the logging to `DEBUG`, I see messages of the form `Unable to add bundle for stage` along with `Stage input watermark: Timestamp(-9223372036854.775000)` (i.e. `timestamp.MIN_TIMESTAMP`) and `Bundle schedule watermark: Timestamp(9223372036854.775000)` (i.e. `timestamp.MAX_TIMESTAMP`) for the two bundles. I have marked this issue as a task because I am not sure if this change in behaviour is by design in version 2.39.0 or if it is actually an issue / bug. Note: I have opened a question [here](https://stackoverflow.com/q/75961407) on this also. ```py import logging import apache_beam as beam def setup_logging(): log_format = '[%(asctime)-15s] [%(name)s] [%(levelname)s]: %(message)s' logging.basicConfig(format=log_format, level=logging.INFO) logging.info("Pipeline Started") class CreateKvPCollectWithSideInputDoFn(beam.DoFn): def __init__(self): super().__init__() def process(self, element, side_input): print(f"side_input_type: {type(side_input)}") yield "b", "2" class CreateKvPCollectDoFn(beam.DoFn): def __init__(self): super().__init__() def process(self, element): yield "a", "1" def main(): setup_logging() pipeline = beam.Pipeline() pcollect_input = ( pipeline | "Input/Create" >> beam.Create(["input"]) ) kvpcollect_1 = ( pcollect_input | "PCollection_1" >> beam.ParDo(CreateKvPCollectDoFn()) ) beamdict_1 = beam.pvalue.AsDict(kvpcollect_1) kvpcollect_2 = ( pcollect_input | "PCollection_2" >> beam.ParDo( CreateKvPCollectWithSideInputDoFn(), side_input=beamdict_1 ) # Commenting in this line allows the example to work on 2.39.0 # | "Reshuffle_2" >> beam.Reshuffle() ) kvpcollect_3 = ( (kvpcollect_1, kvpcollect_2) | "Flatten" >> beam.Flatten() ) beamdict_3 = beam.pvalue.AsDict(kvpcollect_3) ( pcollect_input | "UseBeamDict_3" >> beam.ParDo(CreateKvPCollectWithSideInputDoFn(), side_input=beamdict_3) | "PrintResult" >> beam.Map(print) ) result = pipeline.run() result.wait_until_finish() if __name__ == '__main__': main() ``` ### Issue Priority Priority: 2 (default / most normal work 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 examples - [ ] Component: Beam playground - [ ] Component: Beam katas - [ ] Component: Website - [ ] Component: Spark Runner - [ ] Component: Flink Runner - [ ] Component: Samza 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]
