[Bug]: Periodic sequence never updates the watermark

[damccorm]

What happened?

Right now, if you have a streaming job with periodic impulse as your source, the watermark will not get updated until the periodic impulse completes.

This means that if you have something like:

pipeline
| PeriodicImpulse(time.now(), time.now()+1000000, 1)
| beam.Map(do_stuff)
| beam.WindowInto(window.FixedWindows(2))
| beam.combiners.Count.PerElement()
| beam.Map(log_output)

the pipeline will hang indefinitely and never log anything. If we properly emitted watermarks, it would log output every 2 seconds.

Issue Priority

Priority: 2

Issue Component

Component: sdk-py-core

[Abacn]

Wondering if this is the cause of #22776

[damccorm]

Scratch that (ignore my previous now deleted comment). This seems to fix the problem; I was able to successfully drain a pipeline with my updated periodic sequence code.

[damccorm]

Scratch that - my original analysis was right and we need to explicitly truncate. The drain I thought I saw was actually just my test framework cancelling the job 😳 Explicitly truncating works perfectly though, I'll put up a PR for that shortly.

EDIT - #23765