[FLINK-32414] Watermark alignment will cause flink jobs to hang forever when any source subtask has no SourceSplit

• Type: Sub-task
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.16.2, 1.17.1
• Fix Version/s: 1.18.0, 1.16.3, 1.17.2
• Component/s: Connectors / Common
• Labels: pull-request-available
• Assignee: Rui Fan
• Reporter: Rui Fan
• Created: 22/Jun/23 14:47
• Updated: 06/Jul/23 06:57
• Resolved: 28/Jun/23 02:36

Description

Watermark alignment will cause flink jobs to hang forever when any source subtask has no SourceSplit.

Root cause:

1. SourceOperator#emitLatestWatermark reports the lastEmittedWatermark to SourceCoordinator
2. If one subtask has no SourceSplit, the lastEmittedWatermark will be the Watermark.UNINITIALIZED.getTimestamp() forever, it's Long.MIN_VALUE.
3. SourceCoordinator combines the watermark of all subtasks, and using the minimum watermark as the aggregated watermark.
4. Long.MIN_VALUE must be the minimum watermark, so the maxAllowedWatermark =  Long.MIN_VALUE + maxAllowedWatermarkDrift, and SourceCoordinator will announce it to all subtasks.
5. The maxAllowedWatermark is very small, so all source subtasks will hang forever

How to reproduce?

When the kafka partition number is less than the parallelism of kafka source.

Here is a demo: code link

• kafka partition is 1
• The paralleslism is 2

 

Issue Links

• is caused by: FLINK-24441 Block SourceReader when watermarks are out of alignment
• links to: GitHub Pull Request #22847