Dataflow Job stuck on Draining for long time

[BjornPrime]

What happened?

Issue was reported causing streaming pipeline to get stuck while draining.

Issue was reproduced in the following pipeline:

import logging

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.transforms import trigger
from apache_beam.transforms import window
from apache_beam.transforms.periodicsequence import PeriodicImpulse
import time


def run(argv=None):
  options = PipelineOptions()
  pipeline = beam.Pipeline(options=options)
  _ = (
      pipeline
      |  PeriodicImpulse(
          start_timestamp=time.time(), stop_timestamp=time.time()+15,
          fire_interval=1, apply_windowing=False)
      | beam.WindowInto(
            window.GlobalWindows(),
            trigger=trigger.Repeatedly(trigger.AfterCount(4)),
            accumulation_mode=trigger.AccumulationMode.DISCARDING)
      | beam.combiners.Top.Largest(1)
      | beam.ParDo(lambda x: logging.error('element: %s', x))
  )
  pipeline.run().wait_until_finish()

if __name__ == '__main__':
  logging.getLogger().setLevel(logging.INFO)
  run()

Fails both on Dataflow and Flink.

Example:

./flink-1.11.6/bin/start-cluster.sh

docker run --net=host apache/beam_flink1.11_job_server:latest

python -m pimpulse_small  --runner PortableRunner  --job_endpoint="localhost:8099" --environment_type="LOOPBACK"  --streaming
...
merge_accumulators
    assert result_heap is not None and holds_comparables is not None
RuntimeError: AssertionError [while running 'Largest(1)/Top(1)/CombineGlobally(TopCombineFn)/CombinePerKey/Combine/ParDo(CombineValuesDoFn)']

Issue Priority

Priority: 2

Issue Component

Component: sdk-py-core

[Abacn]

Is this still a problem after #23765 ?

[tvalentyn]

I don't know, I'd suggest try it out and answer here.

[liferoad]

I tested the above code using both DirectRunner and DataflowRunner. Both work fine. So #23765 fixed the assert issue. I suggest we should close this issue and track the future work in #24683 to handle inputs from multiple windows simultaneously.

[tvalentyn]

thanks for testing.