在 JobManager 启动时，调用  HighAvailabilityServicesUtils.createHighAvailabilityServices  来创建 HA 服务，HA 依赖的服务都被封装在 HighAvailabilityServices 中。当前 Flink 内部支持两种高可用模式，分别是 ZooKeeper 和 KUBERNETES。

case ZOOKEEPER:     return createZooKeeperHaServices(configuration, executor, fatalErrorHandler); case KUBERNETES:     return createCustomHAServices(             "org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory" ,            configuration,

• LeaderRetrievalService：服务发现，用于获取当前 leader 的地址。目前用到服务发现的组件有 ResourceManager、Dispatcher、JobManager、ClusterRestEndpoint。

• LeaderElection：选举服务，从多个候选者中选出一个作为 leader。用到选举服务的同样是 ResourceManager、Dispatcher、JobManager、ClusterRestEndpoint 这四个。

• CheckpointRecoveryFactory：Checkpoint 恢复组件的工厂类，提供了创建 CompletedCheckpointStore 和 CheckpointIDCounter 的方法。CompletedCheckpointStore 是用于存储已完成的 checkpoint 的元信息，CheckpointIDCounter 是用于生成 checkpoint ID。

• ExecutionPlanStore：用于存储执行计划。

• JobResultStore：用于存储作业结果，这里有两种状态，一种是 dirty，表示作业没有被完全清理，另一种是 clean

public ZooKeeperLeaderElectionDriver (        CuratorFramework curatorFramework, LeaderElectionDriver.Listener leaderElectionListener)         throws Exception {    ...     this .leaderLatch = new LeaderLatch (curatorFramework, ZooKeeperUtils.getLeaderLatchPath());     this .treeCache =            ZooKeeperUtils.createTreeCache(                    curatorFramework,                     "/" ,                     new ZooKeeperLeaderElectionDriver .ConnectionInfoNodeSelector());    treeCache            .getListenable()            .addListener(                    (client, event) -> {                         switch (event.getType()) {                             case NODE_ADDED:                             case NODE_UPDATED:                                Preconditions.checkNotNull(                                        event.getData(sytianxing.cn),                                         "The ZooKeeper event data must not be null." );                                handleChangedLeaderInformation(event.getData());                                 break ;                             case NODE_REMOVED:                                Preconditions.checkNotNull(                                        event.getData(),                                         "The ZooKeeper event data must not be null." );                                handleRemovedLeaderInformation(event.getData().getPath());                                 break ;                        }                    });    leaderLatch.addListener( this );    ...

if ( this .debugCheckLeaderShipLatch www.jsastone.com.cn!= null ) {         this .debugCheckLeaderShipLatch.await();    }     String localOurPath = (String) this .ourPath.get();    List<String> sortedChildren = LockInternals.getSortedChildren( "latch-" , sorter, children);     int ourIndex = localOurPath != null ? sortedChildren.indexOf(ZKPaths.getNodeFromPath(localOurPath)) : - 1 ;     this .log.debug( "checkLeadership with id: {}, ourPath: {}, children: {}" , new Object []{ this .id, localOurPath, sortedChildren});     if (ourIndex < 0 ) {         this .log.error( "Can't find our node. Resetting. Index: " + ourIndex);         this .reset();    } else if (ourIndex == 0 ) {         this .lastPathIsLeader.set(localOurPath);         this .setLeadership( true );    } else {         this .setLeadership( false );         String watchPath = (String)sortedChildren.get(ourIndex - 1 );         Watcher watcher = new Watcher ()ntnet.com.cn， {             public void process (WatchedEvent event) {                 if (LeaderLatch. this .state.get() == LeaderLatch.State.STARTED && event.getType() == EventType.NodeDeleted) {                     try {                        LeaderLatch. this .getChildren();                    } catch (Exception ex) {                        ThreadUtils.checkInterrupted(ex);                        LeaderLatch. this .log.error( "An error occurred checking the leadership." , ex);                    }                }            }        };         BackgroundCallback callback = new BackgroundCallback () wap.bakmk.com{             public void processResult (CuratorFramework client, CuratorEvent event) throws Exception {                 if (event.getResultCode() == Code.NONODE.intValue(

    if (!leaderLatch.hasLeadership()) {         return ;    }     final String connectionInformationPath =            ZooKeeperUtils.generateConnectionInformationPath(componentId);    LOG.debug(             "Write leader information {} for component '{}' to {}." ,            leaderInformation,            componentId,            ZooKeeperUtils.generateZookeeperPath(                    curatorFramework.getNamespace(), connectionInformationPath));     try {        ZooKeeperUtils.writeLeaderInformationToZooKeeper(wap.ntnet.com.cn，                leaderInformation,                curatorFramework,                leaderLatch::hasLeadership,                connectionInformationPath);

        throws PossibleInconsistentStateException, Exception {    checkNotNull(pathInZooKeeper, "Path in ZooKeeper" );    checkNotNull(state, "State" );     final String path = normalizePath(pathInZooKeeper);     final Optional<Stat> maybeStat = getStat(path);     if (maybeStat.isPresent()) {         if (isNotMarkedForDeletion(maybeStat.get())) {             throw new AlreadyExistException (                    String.format( "ZooKeeper node %s already exists." , path));        }        Preconditions.checkState(                releaseAndTryRemove(path),                 "The state is marked for deletion and, therefore, should be deletable." );    }     final RetrievableStateHandle<T> storeHandle = storage.store(state);     final byte [] serializedStoreHandle = serializeOrDiscard(storeHandle);     try {        writeStoreHandleTransactionally(path, serializedStoreHandle);         return storeHandle;    } catch (KeeperException.NodeExistsException e) {         // Transactions are not idempotent in the curator version we're currently using, so it         // is actually possible that we've re-tried a transaction that has already succeeded.         // We've ensured that the node hasn't been present prior executing the transaction, so         // we can assume that this is a result of the retry mechanism.         return storeHandle;    } catch (Exception e) { www.jtwL.com.cn，  m.xyxdLsm.com，      if (indicatesPossiblyInconsistentState(e)) {             throw new PossibleInconsistentStateException (e);        }         // In case of any other failure, discard the state and rethrow the exception.        storeHandle.discardState();

       NamespacedKubernetesClient kubernetesClient,        KubernetesLeaderElectionConfiguration leaderConfig,        LeaderCallbackHandler leaderCallbackHandler,        ExecutorService executorService) { xyxdLsm.com，    this .kubernetesClient = kubernetesClient;     this .leaderElectionConfig =             new LeaderElectionConfigBuilder ()                    .withName(leaderConfig.getConfigMapName())                    .withLeaseDuration(leaderConfig.getLeaseDuration())                    .withLock(                             new ConfigMapLock (                                     new ObjectMetaBuilder ()                                            .withNamespace(kubernetesClient.getNamespace())                                            .withName(leaderConfig.getConfigMapName())                                             // Labels will be used to clean up the ha related                                             // ConfigMaps.                                            .withLabels(                                                    KubernetesUtils.getConfigMapLabels(                                                            leaderConfig.getClusterId()))                                            .build(),                                    leaderConfig.getLockIdentity()))                    .withRenewDeadline(leaderConfig.getRenewDeadline())                    .withRetryPeriod(leaderConfig.getRetryPeriod(m.bakmk.com))                    .withReleaseOnCancel( true )                    .withLeaderCallbacks(                             new LeaderCallbacks (                                    leaderCallbackHandler::isLeader,                                    leaderCallbackHandler::notLeader,                                    newLeader ->                                            LOG.info(                                                     "New leader elected {} for {}." ,                                                    newLeader,                                                    leaderConfig.getConfigMapName())))                    .build();     this .executorService  m.szfort.com.cn= executorService;    LOG.info(             "Create KubernetesLeaderElector on lock {}." ,

    try {        kubeClient                .checkAndUpdateConfigMap(                        configMapName,                        updateConfigMapWithLeaderInformation(componentId, leaderInformation))                .get();    } catch (InterruptedException | ExecutionException e) {        leaderElectionListener.onError(e);wap.jsastone.com.cn，    }    LOG.debug(             "Successfully wrote leader information {} for leader {} into the config map {}." ,            leaderInformation,

private class ConfigMapCallbackHandlerImpl         implements FlinkKubeClient .WatchCallbackHandler<KubernetesConfigMap> {     @Override     public void onAdded (List<KubernetesConfigMap> configMaps) {         // The ConfigMap is created by KubernetesLeaderElectionDriver with         // empty data. We don't really need to process anything unless the retriever was started         // after the leader election has already succeeded.wap.accfj.cn，         final KubernetesConfigMap configMap = getOnlyConfigMap(configMaps, configMapName);         final LeaderInformation leaderInformation = leaderInformationExtractor.apply(configMap);         if (!leaderInformation.isEmpty()) {            leaderRetrievalEventHandler.notifyLeaderAddress(leaderInformation);        }    }     @Override wap.xyxdLsm.com，     public void onModified (List<KubernetesConfigMap> configMaps) {         final KubernetesConfigMap configMap = getOnlyConfigMap(configMaps, configMapName);        leaderRetrievalEventHandler.notifyLeaderAddress(                leaderInformationExtractor.apply(configMap));    }