Flink kafka Consumer的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从最早的位置开始读取。