深度了解flink(十) JobManager(4) ResourceManager HA

🕗 发布于 2024-11-05 02:45 flink 大数据

ResourceManager（ZK模式）的高可用启动流程

ResourceManager启动流程在DefaultDispatcherResourceManagerComponentFactory#create中

public DispatcherResourceManagerComponent create(
            Configuration configuration,
            ResourceID resourceId,
            Executor ioExecutor,
            RpcService rpcService,
            HighAvailabilityServices highAvailabilityServices,
            BlobServer blobServer,
            HeartbeatServices heartbeatServices,
            DelegationTokenManager delegationTokenManager,
            MetricRegistry metricRegistry,
            ExecutionGraphInfoStore executionGraphInfoStore,
            MetricQueryServiceRetriever metricQueryServiceRetriever,
            Collection<FailureEnricher> failureEnrichers,
            FatalErrorHandler fatalErrorHandler)
            throws Exception {
        //resourcemanager的选举服务
        LeaderRetrievalService resourceManagerRetrievalService = null;
        ResourceManagerService resourceManagerService = null;

        try {
            //resourceManager leader 获取服务
            resourceManagerRetrievalService =
                    highAvailabilityServices.getResourceManagerLeaderRetriever();
            //LeaderGatewayRetriever实现了LeaderRetrievalListener接口
            final LeaderGatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever =
                    new RpcGatewayRetriever<>(
                            rpcService,
                            ResourceManagerGateway.class,
                            ResourceManagerId::fromUuid,
                            new ExponentialBackoffRetryStrategy(
                                    12, Duration.ofMillis(10), Duration.ofMillis(50)));

            
            resourceManagerService =
                    ResourceManagerServiceImpl.create(
                            resourceManagerFactory,
                            configuration,
                            resourceId,
                            rpcService,
                            highAvailabilityServices,
                            heartbeatServices,
                            delegationTokenManager,
                            fatalErrorHandler,
                            new ClusterInformation(hostname, blobServer.getPort()),
                            webMonitorEndpoint.getRestBaseUrl(),
                            metricRegistry,
                            hostname,
                            ioExecutor);

            //启动resourceManagerService start方法会调用具体的选举方法
            resourceManagerService.start();
            //启动etrievalService服务的start方法，
            resourceManagerRetrievalService.start(resourceManagerGatewayRetriever);

            return new DispatcherResourceManagerComponent(
                    dispatcherRunner,
                    resourceManagerService,
                    dispatcherLeaderRetrievalService,
                    resourceManagerRetrievalService,
                    webMonitorEndpoint,
                    fatalErrorHandler,
                    dispatcherOperationCaches);
        } catch (Exception exception) {
            //省略无关代码
        }
    }

初始化LeaderRetrievalServer

resourceManagerRetrievalService =
                    highAvailabilityServices.getResourceManagerLeaderRetriever();

getResourceManagerLeaderRetriever会调用父类AbstractHaServices的这个方法

 @Override
    public LeaderRetrievalService getResourceManagerLeaderRetriever() {
        return createLeaderRetrievalService(getLeaderPathForResourceManager());
    }

createLeaderRetrievalService会调用具体子类的方法，zk模式下会走到ZooKeeperLeaderElectionHaServices下

    @Override
    protected LeaderRetrievalService createLeaderRetrievalService(String componentId) {
        // Maybe use a single service for leader retrieval
        return ZooKeeperUtils.createLeaderRetrievalService(
                curatorFrameworkWrapper.asCuratorFramework(),
                ZooKeeperUtils.getLeaderPath(componentId),
                configuration);
    }

继续方法跟进

public static DefaultLeaderRetrievalService createLeaderRetrievalService(
        final CuratorFramework client, final String path, final Configuration configuration) {
    return new DefaultLeaderRetrievalService(
            createLeaderRetrievalDriverFactory(client, path, configuration));

最终调用DefaultLeaderRetrievalService的构造方法进行初始化，入参是一个LeaderRetrievalDriverFactory接口，zk模式下是ZooKeeperLeaderRetrievalDriverFactory

初始化LeaderRetrievalListener

            final LeaderGatewayRetriever<DispatcherGateway> dispatcherGatewayRetriever =
                    new RpcGatewayRetriever<>(
                            rpcService,
                            DispatcherGateway.class,
                            DispatcherId::fromUuid,
                            new ExponentialBackoffRetryStrategy(
                                    12, Duration.ofMillis(10), Duration.ofMillis(50)));

RpcGatewayRetrieverUML类图如下

LeaderGatewayRetriever#notifyNewLeaderAddress对LeaderRetrievalListener#notifyLeaderAddress做了实现

    @Override
    public void notifyNewLeaderAddress(
            CompletableFuture<Tuple2<String, UUID>> newLeaderAddressFuture) {
        final CompletableFuture<T> newGatewayFuture = createGateway(newLeaderAddressFuture);

        final CompletableFuture<T> oldGatewayFuture =
                atomicGatewayFuture.getAndSet(newGatewayFuture);

        newGatewayFuture.whenComplete(
                (t, throwable) -> {
                    if (throwable != null) {
                        oldGatewayFuture.completeExceptionally(throwable);
                    } else {
                        oldGatewayFuture.complete(t);
                    }
                });
    }

创建ResourceManagerService

 resourceManagerService =
                    ResourceManagerServiceImpl.create(
                            resourceManagerFactory,
                            configuration,
                            resourceId,
                            rpcService,
                            highAvailabilityServices,
                            heartbeatServices,
                            delegationTokenManager,
                            fatalErrorHandler,
                            new ClusterInformation(hostname, blobServer.getPort()),
                            webMonitorEndpoint.getRestBaseUrl(),
                            metricRegistry,
                            hostname,
                            ioExecutor);

ResourceManagerService负责管理ResourceManager的生命周期，也负责haServer的启动

创建LeaderElection

    private ResourceManagerServiceImpl(
            ResourceManagerFactory<?> resourceManagerFactory,
            ResourceManagerProcessContext rmProcessContext)
            throws Exception {
        //省略其他方法
        //LeaderElection进行了初始化
        this.leaderElection =
                rmProcessContext.getHighAvailabilityServices().getResourceManagerLeaderElection();

LeaderElection选举

 resourceManagerService.start();

跳转到start方法

  public void start() throws Exception {
        synchronized (lock) {
            if (running) {
                LOG.debug("Resource manager service has already started.");
                return;
            }
            running = true;
        }

        LOG.info("Starting resource manager service.");
        
        leaderElection.startLeaderElection(this);
    }

leaderElection.startLeaderElection(this)zk模式下走的是DefaultLeaderElection的方法，

    @Override
    public void startLeaderElection(LeaderContender contender) throws Exception {
        Preconditions.checkNotNull(contender);
        parentService.register(componentId, contender);
    }

开始对参选者进行注册

protected void register(String componentId, LeaderContender contender) throws Exception {
        checkNotNull(componentId, "componentId must not be null.");
        checkNotNull(contender, "Contender must not be null.");

        synchronized (lock) {
           
            if (leaderElectionDriver == null) {
                createLeaderElectionDriver();
            }

        //省略无关代码
    }

register会判断选举的driver是否存在，如果不存在，则根据高可用的模式进行选举驱动的创建

 @Override
    public ZooKeeperLeaderElectionDriver create(
            LeaderElectionDriver.Listener leaderElectionListener) throws Exception {
        return new ZooKeeperLeaderElectionDriver(curatorFramework, leaderElectionListener);
    }

ZooKeeperLeaderElectionDriver初始化

public ZooKeeperLeaderElectionDriver(
            CuratorFramework curatorFramework, LeaderElectionDriver.Listener leaderElectionListener)
            throws Exception {
        //参数校验
        this.curatorFramework = Preconditions.checkNotNull(curatorFramework);
        this.leaderElectionListener = Preconditions.checkNotNull(leaderElectionListener);

        //使用ZooKeeperUtils.generateLeaderLatchPath方法基于curatorFramework的命名空间生成一个ZooKeeper节点路径，该路径通常用于领导者选举的锁。
        this.leaderLatchPath =
                ZooKeeperUtils.generateLeaderLatchPath(curatorFramework.getNamespace());
        //使用CuratorFramework和之前生成的路径创建一个LeaderLatch实例。LeaderLatch是Curator提供的一个领导者选举实现。    
        this.leaderLatch = new LeaderLatch(curatorFramework, ZooKeeperUtils.getLeaderLatchPath());
        //使用ZooKeeperUtils.createTreeCache方法创建一个TreeCache实例，用于监听ZooKeeper中特定路径（这里是根路径"/"）下的节点变化
        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(),
                                            "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);
        curatorFramework.getConnectionStateListenable().addListener(listener);
        leaderLatch.start();
        treeCache.start();
    }

ZooKeeperLeaderElectionDriver.handleStateChange

状态变化时候根据不同状态打印日志

private void handleStateChange(ConnectionState newState) {
        switch (newState) {
            case CONNECTED:
                LOG.debug("Connected to ZooKeeper quorum. Leader election can start.");
                break;
            case SUSPENDED:
                LOG.warn("Connection to ZooKeeper suspended, waiting for reconnection.");
                break;
            case RECONNECTED:
                LOG.info(
                        "Connection to ZooKeeper was reconnected. Leader election can be restarted.");
                break;
            case LOST:
                // Maybe we have to throw an exception here to terminate the JobManager
                LOG.warn(
                        "Connection to ZooKeeper lost. None of the contenders participates in the leader election anymore.");
                break;
        }
    }

DefaultLeaderElectionService.notifyLeaderContenderOfLeadership

    private void notifyLeaderContenderOfLeadership(String componentId, UUID sessionID) {
        if (!leaderContenderRegistry.containsKey(componentId)) {
            LOG.debug(
                    "The grant leadership notification for session ID {} is not forwarded because the DefaultLeaderElectionService ({}) has no contender registered.",
                    sessionID,
                    leaderElectionDriver);
            return;
        } else if (!sessionID.equals(issuedLeaderSessionID)) {
            LOG.debug(
                    "An out-dated leadership-acquired event with session ID {} was triggered. The current leader session ID is {}. The event will be ignored.",
                    sessionID,
                    issuedLeaderSessionID);
            return;
        }

        Preconditions.checkState(
                !confirmedLeaderInformation.hasLeaderInformation(componentId),
                "The leadership should have been granted while not having the leadership acquired.");

        LOG.debug(
                "Granting leadership to the contender registered under component '{}' with session ID {}.",
                componentId,
                issuedLeaderSessionID);

        leaderContenderRegistry.get(componentId).grantLeadership(issuedLeaderSessionID);
    }

原文地址：https://blog.csdn.net/qq_40689430/article/details/143418429

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：软件对象粒度控制与设计模式在其中作用的例子
下一篇：基于AI大模型的复杂扫描件PDF信息提取与规整

关于selenium元素找不到的问题（Unable to locate element: {“method“:“xpath“,“selector“:“）
过了两周，查看日志发现，裂开了，又不好使了，然后，我重新运行了一遍，想找找问题，结果他又好使了。我真的裂开，裂开了啊，再然后下载页面，发现他会出现两种页面，并且页面源代码完全不同？之前一个任务让用se
阅读更多2024-11-17
iOS 键盘弹出视图精准上移
注意，我们可以自行设置一个适当的buffer,我这里是40，就是输入框距离。1 视图精准位移，需要计算出输入框距离屏幕底部的距离，然后。计算出输入框需要上移的距离，就是整个视图需要上移的距离,键盘有一
阅读更多2024-11-17
Intern大模型训练营（八）：Llamaindex RAG 实践
首先在Intern Studio中申请30% A100的开发机。进入开发机后，创建新的conda环境，命名为llamaindex复制完成后，在本地查看环境。运行conda命令，激活llamaindex
阅读更多2024-11-17
Spring事务管理
Spring事务管理是Spring框架中的一个重要功能，用于管理数据库事务。它提供了一种声明式的方式来管理事务，简化了开发人员的工作。
阅读更多2024-11-17
介绍一下整数在内存的储存形式(c基础)
把整数以二进制形式写出在前面补零(保证32位)把整数以二进制形式写出在前面补零（第一位为符号位负数为1，正数为0。第一位（左）为符号位补1。（即1变为0，0变为1)
阅读更多2024-11-17
Linux驱动开发第2步_“物理内存”和“虚拟内存”的映射
“新字符设备的GPIO驱动”和“设备树下的GPIO驱动”都要用到寄存器地址，使用“物理内存”和“虚拟内存”映射时，非常不方便，而pinctrl和gpio子系统的GPIO驱动，非常简化。因此，要重点学习
阅读更多2024-11-17
LLMs 激活函数篇
LLMs 激活函数
阅读更多2024-11-17
学习python的第十天之数据类型——dict字典
Python 中的字典（Dictionary）是一个非常强大的内置数据类型，它用来存储键值对（key-value pairs）信息。字典是无序的，这意味着它们不会记录你添加键值对的顺序；然而，从 Py
阅读更多2024-11-17
基于Java Springboot校园共享单车
为了节省时间和提高工作效率，越来越多的人选择利用互联网进行线上打理各种事务，然后线上管理系统也就相继涌现。基于这种情况，我们需要这样一个界面简单大方、功能齐全的系统来解决用户问题，满足用户需求。课题主
阅读更多2024-11-17
HashMap面试知识点
HashMap链表的插入的方式是是头插法，在多线程的情况下，容易产生环形链表，查询时就会产生死循环问题。HashMap的插入法改为了尾插法，但是多线程情况下依然会产生一些问题，例如前面说到的put()
阅读更多2024-11-17