Flink 高可用原理
Flink JobManager 高可用 加强了 Flink 集群防止 JobManager 故障的能力。 此特性确保了 Flink 集群将始终持续执行你提交的作业。
JobManager 高可用一般概念是指，在任何时候都有 一个领导者 JobManager，如果领导者出现故障，则有多个备用 JobManager 来接管领导。这保证了 不存在单点故障，只要有备用 JobManager 担任领导者，程序就可以继续运行。
Flink 的高可用服务封装了所需的服务，使一切可以正常工作：
- 领导者选举：从
n个候选者中选出一个领导者 - 服务发现：检索当前领导者的地址
- 状态持久化：继承程序恢复作业所需的持久化状态（JobGraphs、用户代码jar、已完成的检查点）
为了恢复提交的作业，Flink 会持久化元数据和 job 组件。高可用数据将一直保存，直到相应的作业执行成功、被取消或最终失败。当这些情况发生时，将删除所有高可用数据，包括存储在高可用服务中的元数据。
Flink 组件的高可用通过高可用服务选举组件实现的 LeaderContender 实现主从节点的选举，当节点成为 Leader 节点后再启动相关的组件服务，所以使用高可用服务的组件类型是可以通过 LeaderContender 的子类来查看的。
客户端通过监听高可用服务中节点地址信息的变化，再做出相关改变。最常见的 RestClusterClient，它在创建时就会同时创建LeaderRetrievalListener，当它去请求相关组件时，都会从 LeaderRetrievalListener 中获取监听的地址信息再对服务端进行请求。
需要选举的组件作为 LeaderContender 被注入到 LeaderElectionService 中，LeaderElectionService 将作为选举者参加高可用Leader 的选举。
Flink 将在启动时创建 HighAvailabilityServices，HighAvailabilityServices 可以快速的创建 LeaderElectionService 和 LeaderRetrievalService，HighAvailabilityServices 的好处在于所有的选举服务和组件可以共享同一个高可用服务客户端，以及一些资源的清理和关闭。
Zookeeper 高可用
选举和监听
Flink 的 Zookeeper 高可用是通过 Zookeeper 的 LeaderLatch 来保证的，每个节点组件注册一个 LeaderLatch，Zookeeper 内部将会选举节点成为 Leader。
在高可用服务中， Flink 中的 LeaderElectionDriver 将用于与高可用服务通信，LeaderElectionDriver 一般作为竞选组件参加选举，例如当使用 Zookeeper 高可用服务时中，将使用 ZooKeeperLeaderElectionDriver 实现 LeaderElectionDriver 与 curator 框架中的 LeaderLatchListener，它在创建时同时会创建 CuratorFramework 的 LeaderLatch 参加选举。
LeaderElectionDriver 中有个 LeaderElectionDriver.Listener， 它一般用于监听 LeaderElectionDriver 是否成功成为 Leader。在开启高可用的 Flink 中，DefaultLeaderElectionService 一般作为 LeaderElectionDriver.Listener 的实现类，同时也是 LeaderElectionService 的实现类，LeaderElectionService 前面也讲过了，用于调用 LeaderContender 高可用组件实现类。
具体的高可用组件实现类有：
- WebMonitorEndpoint：Rest 服务器
- MiniDispatcherRestEndpoint
- DispatcherRestEndpoint
- JobMasterServiceLeadershipRunner： 用于启动 JobManager，具体可见[ JobManager创建和启动解析](# JobManager创建和启动解析)
- ResourceManagerServiceImpl：用于启动 ResourceManager，具体可见[ResourceManager创建和启动解析](# ResourceManager创建和启动解析)
- DefaultDispatcherRunner：用于启动 Dispatcher，具体可见[ Dispatcher 创建和启动解析](# Dispatcher 创建和启动解析)
每个高可用组件都将在 Zookeeper 上注册相应的节点进行 Leader 竞选，相关的组件路径如下。
/flink
+/cluster_id_1/leader/latch
| | /resource_manager/connection_info
| | /dispatcher/connection_info
| | /rest_server/connection_info
| | /job-id-1/connection_info
| | /job-id-2/connection_info
| |
| |
| +jobgraphs/job-id-1
| | /job-id-2
| +jobs/job-id-1/checkpoints/latest
| | /latest-1
| | /latest-2
| | /checkpoint_id_counter
ZooKeeperMultipleComponentLeaderElectionHaServices 在启动时创建，作为管理整个 JobManager 高可用的组件对象。
下面解析组件选举 Leader 和 Leader 监听的整个源码流程：
组件选举流程:
-
组件开始选举 Leader。
-
创建Leader竞争者。
new LeaderContender -
通过 leaderName 从高可用服务中获取 DefaultLeaderElectionService 选举服务，leaderName 最后会作为 Zookeeper 节点路径。
ZooKeeperMultipleComponentLeaderElectionHaServices#createLeaderElectionService-
获取或创建单例的LeaderElectionService
ZooKeeperMultipleComponentLeaderElectionHaServices#getOrInitializeSingleLeaderElectionServicenew DefaultMultipleComponentLeaderElectionService
-
-
DefaultLeaderElectionService 为 LeaderContender 竞争 Leader
DefaultLeaderElectionService#start(LeaderContender) -
DefaultLeaderElectionService 启动和判断是否为 Leader 时都会开始竞争 Leader
DefaultLeaderElectionService#onGrantLeadership-
内部的 LeaderContender 开始竞争 Leader，这里使用 ResourceManagerServiceImpl 来举例
LeaderContender#onGrantLeadership-
当前节点的 ResourceManagerServiceImpl 成为 Leader
ResourceManagerServiceImpl#grantLeadership-
创建 ResourceManager
ResourceManagerServiceImpl#createResourceManager -
启动ResourceManager
ResourceManagerServiceImpl#startResourceManagerIfIsLeader -
如果启动完后没有问题，那么确定ResourceManager为Leader
DefaultLeaderElectionService#confirmLeadership-
验证当前 ResourceManagerServiceImpl 是否为Leader
-
如果是 Leader，则创建 LeaderInformation Leader 信息，并写入 Zookeeper 中
MultipleComponentLeaderElectionDriverAdapter#writeLeaderInformation-
发布 LeaderInformation
DefaultMultipleComponentLeaderElectionService#publishLeaderInformationZooKeeperMultipleComponentLeaderElectionDriver#publishLeaderInformation-
生成Zookeeper连接信息路径，ResourceManager的信息就是resource_manager/connection_info，默认为ZooKeeperUtils#getComponentPath/connection_info
ZooKeeperUtils#generateConnectionInformationPath -
将Leader信息写到 Zookeeper 节点中
ZooKeeperUtils#writeLeaderInformationToZooKeeper
-
-
-
-
-
-
-
JobManager Leader 监听流程
-
根据 high-availability.cluster-id 创建 ClientHighAvailabilityServices
ClientHighAvailabilityServicesFactory#create-
根据 Configuration 创建 Zookeeper 的 CuratorFramework，大部分高可用的参数都在这里面使用到
ZooKeeperUtils#startCuratorFramework -
创建ZooKeeperClientHAServices
ZooKeeperClientHAServices#init- 缓存创建的 CuratorFramework 和 Configuration，用于创建 LeaderRetrievalService
-
-
通过高可用服务创建 Leader 监听服务 LeaderRetrievalService
ClientHighAvailabilityServices#getClusterRestEndpointLeaderRetriever-
调用 ZookeeperUtils 创建 DefaultLeaderRetrievalService
ZooKeeperUtils#createLeaderRetrievalService-
创建 ZooKeeperLeaderRetrievalDriverFactory
ZooKeeperUtils#createLeaderRetrievalDriverFactory -
使用 ZooKeeperLeaderRetrievalDriverFactory 创建 DefaultLeaderRetrievalService
-
-
-
Leader 节点监听服务注册需要的监听器
LeaderRetrievalService#start-
使用 ZooKeeperLeaderRetrievalDriverFactory 创建 ZooKeeperLeaderRetrievalDriver
ZooKeeperLeaderRetrievalDriver#ZooKeeperLeaderRetrievalDriver-
缓存 LeaderRetrievalService
-
创建 Zookeeper 的 TreeCache
-
使用 ZooKeeperLeaderRetrievalDriverFactory 中的 CuratorFramework 和监听路径创建 TreeCache
ZooKeeperUtils#createTreeCache -
TreeCache 添加 ZooKeeperLeaderRetrievalDriver 的 retrieveLeaderInformationFromZooKeeper 作为 Leader 节点数据变化后的回调方法，回调时会解析数据，然后会将数据再发送给 LeaderRetrievalService
ZooKeeperUtils#createTreeCacheListenerTreeCache#getListenable#addListenerZooKeeperLeaderRetrievalDriver#retrieveLeaderInformationFromZooKeeper
-
-
启动 TreeCache
TreeCache#start -
将自身的 handleStateChange 方法注册为 CuratorFramework 连接状态改变后的回调方法
CuratorFramework#getConnectionStateListenable#addListenerZooKeeperLeaderRetrievalDriver#handleStateChange
-
-
-
Zookeeper Leader 节点变化后通知 ZooKeeperLeaderRetrievalDriver 节点地址改变
ZooKeeperLeaderRetrievalDriver#retrieveLeaderInformationFromZooKeeper-
获取监听节点的当前信息
TreeCache#getCurrentData -
如果存在信息，那么读取信息中的新地址和 SessionId
LeaderRetrievalEventHandler#notifyLeaderAddress -
将地址和 SessionId 或者空信息发送给 LeaderRetrievalEventHandler (LeaderRetrievalService)
LeaderRetrievalEventHandler#notifyLeaderAddress-
LeaderRetrievalService 再回调 LeaderRetrievalListener 的 notifyLeaderAddress 方法，将地址和 SessionId 传输过去
LeaderRetrievalListener#notifyLeaderAddressLeaderRetrievalListener 就是各个需要监听 JobManager 地址的客户端监听器
客户端有： ResourceManagerLeaderListener、LeaderRetriever、JobLeaderIdListener、JobManagerLeaderListener、LeaderConnectingAddressListener 和 LeaderInformationListener
-
-
JobManager 结束清理数据
JobManaager 结束后需要清理的 Zookeeper 数据主要有 JobGraph、Job Checkpoint 和 无用的节点
JobGraphStore、CheckpointRecovery
-
集群停止，停止高可用并清理高可用信息
HighAvailabilityServices#closeWithOptionalClean-
清理高可用信息
HighAvailabilityServices#cleanupAllData-
删除当前 JobManager 在 Zookeeper 上创建的节点以及子节点，也就是 /KaTeX parse error: Expected 'EOF', got '}' at position 38: …eeper.path.root}̲/{high-availability.cluster-id} 节点
curatorFramework#delete#idempotent#deletingChildrenIfNeeded#forPath("/"); -
清理无用的 Zookeeper 节点路径
AbstractHaServices#internalCleanup- 获取 NameSpace, 一般为/$high-availability.zookeeper.path.root}
- 开始循环删除 Namespace 节点以及全部空的父节点，如果节点非空就无法删除
-
清理 BlobStoreService
BlobStoreService#cleanupAllData
-
-
关闭服务和客户端
HighAvailabilityServices#close
-
参考
官方Flink 高可用概念：https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/deployment/ha/overview/
