[FLINK-21817] New Kafka Source might break subtask and split assignment upon rescale

• Type: Bug
• Status: Closed
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.12.2
• Fix Version/s: 1.13.0, 1.12.3
• Component/s: Connectors / Common
• Labels: pull-request-available
• Assignee: Qingsheng Ren
• Reporter: Kezhu Wang
• Created: 16/Mar/21 09:16
• Updated: 28/Aug/21 11:13
• Resolved: 08/Apr/21 22:46

Description

On restoring, splits are add back directly to SourceReader in SourceOperator. In no rescaled restoring, bindings between splits and subtasks are preserved due to no repartition in RoundRobinOperatorStateRepartitioner. But in rescaled restoring, these operator states will be redistributed cross existing subtasks. This might break possible assignment from SourceEnumerator.

Given KafkaSource as an example, the partition to subtask assignment is decided by KafkaSourceEnumerator.getSplitOwner. The mappings will break after rescaling.

I pushed a test case using KafkaSource for evaluation.

I think it requires api addition to solve in generic and configurable way.

Is it a valid issue ? I am not that sure.

cc  jqin sewen

Issue Links

• links to: GitHub Pull Request #15397
• links to: GitHub Pull Request #15461