flink如何支持kafka容灾自动切换

背景

在flink消费kafka消息时，我们会指定连接的kafka服务器的地址以及起始消费偏移等信息，一旦指定，当kafka服务器挂掉后，flink也会由于连接不上服务器而导致失败，这里想要解决的问题是当kafka在机房A挂掉后，如果机房B有对kafka进行容灾的频道，那么flink怎么可以做到自动切换到机房B的进行kafka消费？同理，当机房A数据恢复后，如何自动切回到机房A进行消费？这个过程自动发生而不需要手动修改kafka的地址

技术实现

flink消费kafka的实现类是FlinkKafkaConsumerBase，这个类内部有一个功能：可以自动发现满足某个规则的kafka主题并消费，其关键代码如下：

private void createAndStartDiscoveryLoop(AtomicReference<Exception> discoveryLoopErrorRef) {discoveryLoopThread =new Thread(() -> {try {// --------------------- partition discovery loop// ---------------------// throughout the loop, we always eagerly check if we are still// running before// performing the next operation, so that we can escape the loop as// soon as possiblewhile (running) {if (LOG.isDebugEnabled()) {LOG.debug("Consumer subtask {} is trying to discover new partitions ...",getRuntimeContext().getIndexOfThisSubtask());}final List<KafkaTopicPartition> discoveredPartitions;try {discoveredPartitions =partitionDiscoverer.discoverPartitions();} catch (AbstractPartitionDiscoverer.WakeupException| AbstractPartitionDiscoverer.ClosedException e) {// the partition discoverer may have been closed or woken up// before or during the discovery;// this would only happen if the consumer was canceled;// simply escape the loopbreak;}// no need to add the discovered partitions if we were closed// during the meantimeif (running && !discoveredPartitions.isEmpty()) {kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);}// do not waste any time sleeping if we're not running anymoreif (running && discoveryIntervalMillis != 0) {try {Thread.sleep(discoveryIntervalMillis);} catch (InterruptedException iex) {// may be interrupted if the consumer was canceled// midway; simply escape the loopbreak;}}}} catch (Exception e) {discoveryLoopErrorRef.set(e);} finally {// calling cancel will also let the fetcher loop escape// (if not running, cancel() was already called)if (running) {cancel();}}},"Kafka Partition Discovery for "+ getRuntimeContext().getTaskNameWithSubtasks());discoveryLoopThread.start();}

如上所示，他是通过开启一个线程，然后定时检测的方式来发现是否有新的符合规则条件的主题，如果有添加到消费队列中，读者会不会很好奇，我们讨论的是flink如何对kafka进行容灾切换，你和我说这个主题自动发现做什么？
其实这里想表达的是一样的思路，我们进行kafka容灾的切换也是可以这样做，我们开启一个线程，然后线程里面不停的检测当前消费的kafka集群的连通性是否正常，如果连接不上，那么表明发生了kafka的机房容灾，flink需要切换到kafka的机房B进行消费，那么这里剩下的就只是如何确定消费的偏移量的问题了