[FLINK-35157] Sources with watermark alignment get stuck once some subtasks finish - ASF JIRA

Attach files

Attach Screenshot

Voters

Watch issue

Watchers

Create sub-task

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Bug
Status: Resolved
Priority: Critical
Resolution: Fixed
Affects Version/s: 1.17.2, 1.19.0, 1.18.1
Fix Version/s: 1.17.3, 1.18.2, 1.20.0, 1.19.2
Component/s: Runtime / Coordination
Labels:
- pull-request-available

Description

The current watermark alignment logic can easily get stuck if some subtasks finish while others are still running.

The reason is that once a source subtask finishes, the subtask is not excluded from alignment, effectively blocking the rest of the job to make progress beyond last wm + alignment time for the finished sources.

This can be easily reproduced by the following simple pipeline:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(2);
    DataStream<Long> s = env.fromSource(new NumberSequenceSource(0, 100),
            WatermarkStrategy.<Long>forMonotonousTimestamps().withTimestampAssigner((SerializableTimestampAssigner<Long>) (aLong, l) -> aLong).withWatermarkAlignment("g1", Duration.ofMillis(10), Duration.ofSeconds(2)),
            "Sequence Source").filter((FilterFunction<Long>) aLong -> {
        Thread.sleep(200);
        return true;
    }
);

    s.print();
    env.execute();

The solution could be to send out a max watermark event once the sources finish or to exclude them from the source coordinator