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 possible
while (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 loop
break;
}
// no need to add the discovered partitions if we were closed
// during the meantime
if (running && !discoveredPartitions.isEmpty()) {
kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);
}
// do not waste any time sleeping if we're not running anymore
if (running && discoveryIntervalMillis != 0) {
try {
Thread.sleep(discoveryIntervalMillis);
} catch (InterruptedException iex) {
// may be interrupted if the consumer was canceled
// midway; simply escape the loop
break;
}
}
}
} 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进行消费，那么这里剩下的就只是如何确定消费的偏移量的问题了
