|
|
hi，I have a source Kafka and a sink Kafka，when the amount of data processing grows，I need to expand Kafka topic's partition number ,but I don't want to restart the job to take effect.
for source Kafka, I use flink.partition-discovery.interval-millis and it could consume the new parititon after I expand the Kafka topic's partition number.
but sink kafka don't work like this：
The flink kafka producer get topic's paritition list and cache in topicPartitionsMap as showed In Class FlinkKafkaProducer<IN> :
int[] partitions = (int[])this.topicPartitionsMap.get(targetTopic);
if (null == partitions) {
partitions = getPartitionsByTopic(targetTopic, transaction.producer);
this.topicPartitionsMap.put(targetTopic, partitions);
}When kafka topic needs to expand，the new parition can not be discovered。For example, when expanding from 1 partition to 2 partitions，partitions we got is never update until job restartAre there plans to improve this feature or Is there any other way to achieve the function?
|
Locked
