如何正确使用Flink Connector？(11)_本文主要分享Flinkconnector相关内容

topic 和 partition 动态发现

实际的生产环境中可能有这样一些需求，比如场景一，有一个 Flink 作业需要将五份数据聚合到一起，五份数据对应五个 kafka topic ，随着业务增长，新增一类数据，同时新增了一个 kafka topic ，如何在不重启作业的情况下作业自动感知新的 topic 。场景二，作业从一个固定的 kafka topic 读数据，开始该 topic 有 10 个 partition ，但随着业务的增长数据量变大，需要对 kafka partition 个数进行扩容，由 10 个扩容到 20 。该情况下如何在不重启作业情况下动态感知新扩容的 partition？

针对上面的两种场景，首先需要在构建 FlinkKafkaConsumer 时的 properties 中设置 flink.partition-discovery.interval-millis 参数为非负值，表示开启动态发现的开关，以及设置的时间间隔。此时 FlinkKafkaConsumer 内部会启动一个单独的线程定期去 kafka 获取最新的 meta 信息。针对场景一，还需在构建 FlinkKafkaConsumer 时， topic 的描述可以传一个正则表达式描述的 pattern 。每次获取最新 kafka meta 时获取正则匹配的最新 topic 列表。针对场景二，设置前面的动态发现参数，在定期获取 kafka 最新 meta 信息时会匹配新的 partition 。为了保证数据的正确性，新发现的 partition 从最早的位置开始读取。