[Bug]: Impulse in FlinkRunner does not emit watermark

[je-ik]

What happened?

Impulse in Flink creates FlinkBoundedSource and does not emit watermarks until --shutdownSourcesAfterIdleMs time elapses (default is infinite).

Related issues is that if job fails and is restored before the Impulse reader is shutdown, the impulse is resent, which might retrigger recomputation downstream.

Issue Priority

Priority: 3 (minor)

Issue Components

• 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: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[je-ik]

@becketqin
Hi, this seems to be related to #25525, I wonder what is the best fix path would be. I think it would make sense to use FlinkBoundedSource only for bounded Pipelines and FlinkUnboundedSource for all other cases. This will need tweaking the source to be able to accept BoundedSource.

Or would you see some other option? I tried to fix FlinkBoundedSource to emit MAX_WATERMARK when there is no more splits, but there is still issue with checkpointing and resending the data (if the fail occurs before the source is shutdown).