[FLINK-22147] Refactor Partition Discovery Logic in KafkaSourceEnumerator

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Implemented
• Affects Version/s: 1.13.0
• Fix Version/s: 1.14.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Qingsheng Ren
• Reporter: Qingsheng Ren
• Created: 08/Apr/21 03:48
• Updated: 25/Jun/21 05:18
• Resolved: 25/Jun/21 05:18

Description

Currently the logic of partition discovery is: the worker thread checks if there's new partitions and initialize new splits if so, then coordinator thread marks these splits as pending and try to make assignments.

Under current design, the worker thread needs to keep an internal data structure tracking already discovered partitions, which is duplicated with pending splits + assigned partitions tracked by coordinator thread. Usually this kind of double-bookkeeping is fragile. 

Another issue is that the worker thread always fetches descriptions of ALL topics at partition discovery, which will comes to a problem working with a giant Kafka clusters with millions of topics/partitions. 

In order to fix issues above, a refactor is needed for the partition discovery logic in Kafka enumerator. Basically the logic can be changed to:

1. The worker thread fetches descriptions of subscribed topics/partitions, then hands over to coordinator thread
2. The coordinator thread filters out already discovered partitions (pending + assigned partitions), then invokes worker thread with callAsync to fetch offsets for new partitions
3.  The worker thread fetches offsets and creates splits for new partitions, then hands over new splits to coordinator thread
4. The coordinator thread marks these splits as pending and try to make assignment. 

Discussion of this issue can be found in https://github.com/apache/flink/pull/15461 .

Issue Links

• links to: GitHub Pull Request #15531