[FLINK-26938] HybridSource recovery from savepoint fails When flink parallelism is greater than the number of Kafka partitions

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.14.0
• Fix Version/s: 1.16.0
• Component/s: API / DataStream

Environment

Flink 1.14.0

• Assignee: Nicholas Jiang
• Reporter: wenbao
• Created: 31/Mar/22 05:34
• Updated: 15/Aug/22 03:25
• Resolved: 15/Aug/22 03:25

Description

HybridSource recovery from savepoint fails When flink parallelism is greater than the number of Kafka partitions

First test

Flink job before savePoint
    flink parallelism =16
    kafka partition=3
Flink after savePoint
case 1：
    flink parallelism =16
    kafka partition=3

HybridSource recovery from savepoint fails 

case 2：
    flink parallelism =3
    kafka partition=3
HybridSource recovery from savepoint  successful

case 3：
    flink parallelism =8
    kafka partition=3
HybridSource recovery from savepoint fails  the same NullPointerException: Source for index=0 not available

case 4：
    flink parallelism =4
    kafka partition=3
HybridSource recovery from savepoint fails  the same NullPointerException: Source for index=0 not available

case 5：
    flink parallelism =1
    kafka partition=3
HybridSource recovery from savepoint  successful

Second test

Flink job before savePoint
    flink parallelism =3
    kafka partition=3
Flink after savePoint
case 1：
    flink parallelism =3
    kafka partition=3
HybridSource recovery from savepoint  successful

case 2：
    flink parallelism =1
    kafka partition=3
HybridSource recovery from savepoint  successful

case 3：
    flink parallelism =4
    kafka partition=3
HybridSource recovery from savepoint fails  the same NullPointerException: Source for index=0 not available

Specific code see the attached test code HybridSourceTest