[Task]: Beam Python Version > 2.38.0 DirectRunner ~ AssertionError: A total of N watermark-pending bundles did not execute

[francisduffy]

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 on this also.

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

• 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

[tvalentyn]

Thanks for opening! Note that you could probably find the offending commit with git bisect, if you are comfortable with running beam from master (s.apache.org/beam-python-dev-wiki has some tips on that)

[tvalentyn]

Quick question: does your experiment use direct runner in Streaming or Batch mode ?

[francisduffy]

Thanks a lot for the tips on tracking down the offending commit. I believe that, bisecting between v2.38.0 and v2.39.0, that the offending commit is 8a7c92c from the pull request #16841.

Quick question: does your experiment use direct runner in Streaming or Batch mode ?

The experiment uses Batch mode.

[tvalentyn]

Thanks. This commit also caused other regressions in direct runner, see: #25315
The original author who was taking a look is currently OOO for a bit, so it may take some time to investigate and fix. But if you or someone else is willing to take a closer look, any help is appreciated.

In the meantime I'll add this to the backlog of Direct runner issues I am aware of for future prioritization.