[FLINK-35631] KafkaSource parameter partition.discovery.interval.ms with a default value of 5 minutes does not take effect

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: kafka-3.1.0
• Component/s: Connectors / Kafka
• Assignee: Unassigned
• Reporter: elon_X
• Created: 18/Jun/24 08:48
• Updated: 19/Jun/24 07:11

Description

When I start a stream program to consume Kafka (flink-connector-kafka-3.1-SNAPSHOT) the Flink task does not automatically detect new partitions after Kafka adds partitions.

 

Reason

In the KafkaSourceBuilder, this parameter is checked to see if it has been overridden. Since I did not set this parameter, even though it is CONTINUOUS_UNBOUNDED, it still sets partition.discovery.interval.ms = -1.

In the KafkaSourceEnumerator, the value of partition.discovery.interval.ms is then -1, instead of the default value of 5 minutes, so automatic partition discovery does not work, and the default value of 5 minutes for partition.discovery.interval.ms is meaningless.

 

A possible solution is to set partition.discovery.interval.ms = -1 only if boundedness == Boundedness.BOUNDED is true.