[FLINK-32020] Enable Dynamic Partition Discovery by Default in Kafka Source based on FLIP-288

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: kafka-3.0.0
• Fix Version/s: kafka-3.1.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Hongshun Wang
• Reporter: Hongshun Wang
• Created: 06/May/23 02:52
• Updated: 26/Jan/24 08:33
• Resolved: 14/Jul/23 06:39

Description

As described in FLIP-288, dynamic partition discovery is disabled by default, and users have to specify the interval of discovery in order to turn it on.

This subtask is to enable Dynamic Partition Discovery by Default in Kafka Source.

Partition discovery is performed on the KafkaSourceEnumerator, which asynchronously fetches topic metadata from the Kafka cluster and checks if there are any new topics and partitions. This should not cause performance issues on the Flink side.

On the Kafka broker side, partition discovery sends a MetadataRequest to the Kafka broker to fetch topic information. Considering that the Kafka broker has its metadata cache and the default request frequency is relatively low (once every 30 seconds), this is not a heavy operation, and the broker's performance will not be significantly affected.

Issue Links

• links to: GitHub Pull Request #40