Flink HA简介

flink HA的源码主要在runtime项目下以下三个包：
org.apache.flink.runtime.highavailability：HA服务公共接口类，主要包括初始化、JobStore、ClientHA等。
org.apache.flink.runtime.leaderelection：lead选举，K8s HA对原理进行了简单的介绍
org.apache.flink.runtime.leaderretrieval：lead获取，通过k8s watch机制监听ConfigMap实现。主要在client角色中使用，如RestClusterClient和WebMonitorEndpoint。

Flink HA目前有两种实现方式：zookeeper和k8s。具体实现代码此处也只是分析k8s实现。

  1. kubernetes.cluster-id: <cluster-id>
  2. high-availability: kubernetes
  3. high-availability.storageDir: hdfs:///flink/recovery
  4. kubernetes.jobmanager.replicas: 3 #默认为1，HA配置时最好设为大于1，以便快速恢复

HA服务加载过程分析

  1. 入口在HighAvailabilityServicesUtils#createHighAvailabilityServices。
    根据不同的high-availability配置参数选择不同的HA实现，可以配置NONE、ZOOKEEPER、KubernetesHaServicesFactory接口实现类三种。

  2. 当配置KubernetesHaServicesFactory时，会去加载类并初始化。k8s只提供了Factory配置方式。

    1. public static HighAvailabilityServices createHighAvailabilityServices(
    2.          Configuration configuration,
    3.          Executor executor,
    4.          AddressResolution addressResolution,
    5.          RpcSystemUtils rpcSystemUtils,
    6.          FatalErrorHandler fatalErrorHandler)
    7.          throws Exception {
    8.  //high-availability根据high-availability获取mode，配置Factory类名时，统一为FACTORY_CLASS mode。
    9.  HighAvailabilityMode highAvailabilityMode = HighAvailabilityMode.fromConfig(configuration);
    11.  switch (highAvailabilityMode) {
    12.      case NONE:
    13.          ...
    14.          return new StandaloneHaServices(
    15.                  resourceManagerRpcUrl, dispatcherRpcUrl, webMonitorAddress);
    16.      case ZOOKEEPER:
    17.          return createZooKeeperHaServices(configuration, executor, fatalErrorHandler);
    19.      case FACTORY_CLASS:
    20.          return createCustomHAServices(configuration, executor);
    22.      default:
    23.          throw new Exception("Recovery mode " + highAvailabilityMode + " is not supported.");
    24.  }
    25. }
    26. //加载Factory
    27. private static HighAvailabilityServicesFactory loadCustomHighAvailabilityServicesFactory(
    28.      String highAvailabilityServicesFactoryClassName) throws FlinkException {
    29.  final ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
    31.  return InstantiationUtil.instantiate(
    32.          highAvailabilityServicesFactoryClassName,
    33.          HighAvailabilityServicesFactory.class,
    34.          classLoader);
    35. }

  3. K8s的配置为org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory，会根据high-availability.use-old-ha-services配置选择是否使用老版HA接口实现。
    主要区别在于LeaderElectionService是每个flink组件一个还是公用一个MultipleComponentLeaderElectionService，下面会具体说明。

    1. public HighAvailabilityServices createHAServices(Configuration configuration, Executor executor)
    2.          throws Exception {
    3.  final boolean useOldHaServices =
    4.          configuration.get(HighAvailabilityOptions.USE_OLD_HA_SERVICES);
    6.  if (useOldHaServices) {
    7.      return new KubernetesHaServices(
    8.              FlinkKubeClientFactory.getInstance()
    9.                      .fromConfiguration(configuration, "kubernetes-ha-services"),
    10.              executor,
    11.              configuration,
    12.              BlobUtils.createBlobStoreFromConfig(configuration));
    13.  } else {
    14.      return new KubernetesMultipleComponentLeaderElectionHaServices(
    15.              FlinkKubeClientFactory.getInstance()
    16.                      .fromConfiguration(configuration, "kubernetes-ha-services"),
    17.              executor,
    18.              configuration,
    19.              BlobUtils.createBlobStoreFromConfig(configuration),
    20.              error ->
    21.                      FatalExitExceptionHandler.INSTANCE.uncaughtException(
    22.                              Thread.currentThread(), error));
    23.  }
    24. }

    Flink leaderelection源码分析

    核心类和接口分析

  • HighAvailabilityServices：提供了leaderelection、leaderretrieval和其他ha服务的初始化。
  • DefaultLeaderElectionService：实现了LeaderElectionService和LeaderElectionEventHandler接口。
  • LeaderElectionEventHandler：接口用来做k8s leaderElection选举回调。
  • LeaderElectionService：接口应用于Dispatch等具体某一个LeaderContender组件来做选举。每个组件都创建一个LeaderElectionLock对象（ConfigMap）
  • MultipleComponentLeaderElectionService：同一节点上的多个组件公用一个LeaderElectionLock（也就是只存在一个ConfigMap）
  • LeaderElectionDriver：LeaderElectionService的成员变量，执行选举操作和存储leader信息。Driver当中调用KubernetesLeaderElector类具体实现选举。
  • LeaderContender：需要lead election的组件要实现接口，LeaderElectionEventHandler回调最终会回调到LeaderContender接口中的方法。

    整个LeaderElection主要三个关键部分：LeaderContender是需要选举功能的组件，LeaderElectionService负责选举整个流程服务，LeaderElectionDriver负责和依赖的三方HA模块具体交互实现。

LeaderContender组件创建&选举

通过Dispatcher启动流程看leaderElection，入口在org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory#create。

  1. dispatcherRunner =
  2. dispatcherRunnerFactory.createDispatcherRunner(
  3.         highAvailabilityServices.getDispatcherLeaderElectionService(),
  4.         fatalErrorHandler,
  5.         new HaServicesJobPersistenceComponentFactory(highAvailabilityServices),
  6.         ioExecutor,
  7.         rpcService,
  8.         partialDispatcherServices);

DispatchRunner作为一个LeaderContender，在创建过程中，首先需要创建DispatcherLeaderElectionService。
由于MultipleComponentLeaderElectionService是公用的，所以highAvailabilityServices里只会初始化一次。

  1. @Override
  2. public LeaderElectionService getDispatcherLeaderElectionService() {
  3.     return createLeaderElectionService(getLeaderPathForDispatcher());
  4. }
  6. protected LeaderElectionService createLeaderElectionService(String leaderName) {
  7.     final MultipleComponentLeaderElectionService multipleComponentLeaderElectionService =
  8.             getOrInitializeSingleLeaderElectionService();
  9.     //此处和非Multiple一样，还是使用DefaultLeaderElectionService。
  10.     //底层是用MultipleComponentLeaderElectionDriverAdapter代理到MultipleComponentLeaderElectionService。
  11.     return new DefaultLeaderElectionService(
  12.             multipleComponentLeaderElectionService.createDriverFactory(leaderName));
  13. }
  15. private DefaultMultipleComponentLeaderElectionService
  16.             getOrInitializeSingleLeaderElectionService() {
  17.     synchronized (lock) {
  18.         //multipleComponentLeaderElectionService初始化一次。
  19.         if (multipleComponentLeaderElectionService == null) {
  20.             try {
  22.                 final KubernetesLeaderElectionConfiguration leaderElectionConfiguration =
  23.                         new KubernetesLeaderElectionConfiguration(
  24.                                 getClusterConfigMap(), lockIdentity, configuration);
  25.                 multipleComponentLeaderElectionService =
  26.                         new DefaultMultipleComponentLeaderElectionService(
  27.                                 fatalErrorHandler,
  28.                                 new KubernetesMultipleComponentLeaderElectionDriverFactory(
  29.                                         kubeClient,
  30.                                         leaderElectionConfiguration,
  31.                                         configMapSharedWatcher,
  32.                                         watchExecutorService,
  33.                                         fatalErrorHandler));
  34.             } catch (Exception e) {
  35.                 throw new FlinkRuntimeException(
  36.                         "Could not initialize the default single leader election service.", e);
  37.             }
  38.         }
  40.         return multipleComponentLeaderElectionService;
  41.     }
  42. }

DispatcherLeaderElectionService创建好后，createDispatcherRunner会创建dispatchRunner，并走到DispatcherRunnerLeaderElectionLifecycleManager构造函数中start DispatcherLeaderElectionService，创建好的dispatchRunner作为DispatcherLeaderElectionService的回调组件，开始选举。

  1. private DispatcherRunnerLeaderElectionLifecycleManager(
  2.             T dispatcherRunner, LeaderElectionService leaderElectionService) throws Exception {
  3.     this.dispatcherRunner = dispatcherRunner;
  4.     this.leaderElectionService = leaderElectionService;
  6.     leaderElectionService.start(dispatcherRunner);
  7. }
  9. //start过程中会注册当前DispatcherLeaderElectionService到DefaultMultipleComponentLeaderElectionService中。
  10. //DefaultMultipleComponentLeaderElectionService使用leaderElectionEventHandlers保存所有注册的LeaderContender。
  11. private final Map<String, LeaderElectionEventHandler> leaderElectionEventHandlers;
  12. //使用KubernetesMultipleComponentLeaderElectionDriver执行具体选举
  13. private final MultipleComponentLeaderElectionDriver multipleComponentLeaderElectionDriver;
  15. public KubernetesMultipleComponentLeaderElectionDriver(
  16.             KubernetesLeaderElectionConfiguration leaderElectionConfiguration,
  17.             FlinkKubeClient kubeClient,
  18.             Listener leaderElectionListener,
  19.             KubernetesConfigMapSharedWatcher configMapSharedWatcher,
  20.             Executor watchExecutor,
  21.             FatalErrorHandler fatalErrorHandler) {
  22.     Preconditions.checkNotNull(leaderElectionConfiguration);
  23.     this.kubeClient = Preconditions.checkNotNull(kubeClient);
  24.     this.leaderElectionListener = Preconditions.checkNotNull(leaderElectionListener);
  25.     this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
  26.     Preconditions.checkNotNull(configMapSharedWatcher);
  27.     Preconditions.checkNotNull(watchExecutor);
  29.     this.configMapName = leaderElectionConfiguration.getConfigMapName();
  30.     this.lockIdentity = leaderElectionConfiguration.getLockIdentity();
  32.     this.leaderElector =
  33.             kubeClient.createLeaderElector(
  34.                     leaderElectionConfiguration, new LeaderCallbackHandlerImpl());
  36.     this.configMapLabels =
  37.             KubernetesUtils.getConfigMapLabels(
  38.                     leaderElectionConfiguration.getClusterId(),
  39.                     LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY);
  41.     kubernetesWatch =
  42.             configMapSharedWatcher.watch(
  43.                     configMapName, new ConfigMapCallbackHandlerImpl(), watchExecutor);
  45.     //leaderElector run启动选举
  46.     leaderElector.run();
  48.     LOG.debug("Starting the {} for config map {}.", getClass().getSimpleName(), configMapName);
  49. }

LeaderContender启动&回写leadAddress

选举成功，执行DefaultMultipleComponentLeaderElectionService回调启动LeaderContender组件。

  1. //DefaultMultipleComponentLeaderElectionService isLeader回调，遍历所有LeaderContender，调用对应的DefaultLeaderElectionService onGrantLeadership回调
  2. public void isLeader() {
  3.     final UUID newLeaderSessionId = UUID.randomUUID();
  4.     synchronized (lock) {
  5.         if (!running) {
  6.             return;
  7.         }
  9.         currentLeaderSessionId = UUID.randomUUID();
  11.         forEachLeaderElectionEventHandler(
  12.                 leaderElectionEventHandler ->
  13.                         leaderElectionEventHandler.onGrantLeadership(newLeaderSessionId));
  14.     }
  15. }
  17. //DefaultLeaderElectionService onGrantLeadership里调用leaderContender.grantLeadership(issuedLeaderSessionID)
  18. //此处选择Dispatch来分析
  19. public void grantLeadership(UUID leaderSessionID) {
  20.     runActionIfRunning(
  21.             () -> {
  22.                 LOG.info(
  23.                         "{} was granted leadership with leader id {}. Creating new {}.",
  24.                         getClass().getSimpleName(),
  25.                         leaderSessionID,
  26.                         DispatcherLeaderProcess.class.getSimpleName());
  27.                 //创建并启动Dispatcher
  28.                 startNewDispatcherLeaderProcess(leaderSessionID);
  29.             });
  30. }
  32. private void startNewDispatcherLeaderProcess(UUID leaderSessionID) {
  33.     stopDispatcherLeaderProcess();
  34.     //创建dispatcherLeaderProcess
  35.     dispatcherLeaderProcess = createNewDispatcherLeaderProcess(leaderSessionID);
  37.     final DispatcherLeaderProcess newDispatcherLeaderProcess = dispatcherLeaderProcess;
  38.     FutureUtils.assertNoException(
  39.             previousDispatcherLeaderProcessTerminationFuture.thenRun(
  40.                     newDispatcherLeaderProcess::start));
  41. }
  43. //startNewDispatcherLeaderProcess-》createNewDispatcherLeaderProcess-》forwardConfirmLeaderSessionFuture
  44. //此处往ConfigMap data里写leaderAddress，对应leaderElectionService.confirmLeadership方法
  45. private void forwardConfirmLeaderSessionFuture(
  46.             UUID leaderSessionID, DispatcherLeaderProcess newDispatcherLeaderProcess) {
  47.     FutureUtils.assertNoException(
  48.             newDispatcherLeaderProcess
  49.                     .getLeaderAddressFuture()
  50.                     .thenAccept(
  51.                             leaderAddress -> {
  52.                                 if (leaderElectionService.hasLeadership(leaderSessionID)) {
  53.                                     leaderElectionService.confirmLeadership(
  54.                                             leaderSessionID, leaderAddress);
  55.                                 }
  56.                             }));
  57. }

Flink leaderretrieval源码分析

核心类和接口分析

  • HighAvailabilityServices：提供了leaderelection、leaderretrieval和其他ha服务的初始化。
  • LeaderRetrievalService：提供lead变动回调服务。
  • LeaderRetrievalListener：获取lead变动的组件需要实现此接口。并作为LeaderRetrievalService start方法的参数，注册到LeaderRetrievalService
  • LeaderRetrievalDriver：具体实现lead变动监听功能。

    源码分析

    具体逻辑和LeadElection类似，变动监听使用K8s watch功能。
    具体逻辑略.
    k8s-client java版code demo
    InformerExample
    WatchExample