Hadoop3.2.1 【 YARN 】源码分析 : ApplicationMaster 浅析

太过爱你忘了你带给我的痛 2022-12-27 14:23 265阅读 0赞

# 一 . 前言 #

ApplictionMaster管理部分主要由三个服务构成， 分别是ApplicationMasterLauncher、AMLivelinessMonitor和ApplicationMasterService， 它们共同管理应用程序的ApplicationMaster的生存周期。

# 二 . 整体流程 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

**步骤1** : 用户向YARN ResourceManager提交应用程序， ResourceManager收到提交请求后， 先向资源调度器申请用以启动ApplicationMaster的资源， 待申请到资源后， 再由ApplicationMasterLauncher与对应的NodeManager通信， 从而启动应用程序的ApplicationMaster。  
**步骤2** : ApplicationMaster启动完成后， ApplicationMasterLauncher会通过事件的形式， 将刚刚启动的ApplicationMaster注册到AMLivelinessMonitor， 以启动心跳监控。  
**步骤3** : ApplicationMaster启动后， 先向ApplicationMasterService注册， 并将自己所在host、 端口号等信息汇报给它。  
**步骤4** : ApplicationMaster运行过程中， 周期性地向ApplicationMasterService汇报“心跳”信息（“心跳”信息中包含想要申请的资源描述） .  
**步骤5** : ApplicationMasterService每次收到ApplicationMaster的心跳信息后， 将通知AMLivelinessMonitor更新该应用程序的最近汇报心跳的时间。  
**步骤6** : 当应用程序运行完成后， ApplicationMaster向ApplicationMasterService发送请求， 注销自己。  
**步骤7** : ApplicationMasterService收到注销请求后， 标注应用程序运行状态为完成， 同时通知AMLivelinessMonitor移除对它的心跳监控

# 三 . ApplicationMasterLauncher #

ApplicationMasterLauncher既是一个服务， 也是一个事件处理器， 它处理AMLauncherEvent类型的事件， 该类型事件有两种， 分别是请求启动一个ApplicationMaster的"LAUNCH"事件和请求清理一个ApplicationMaster的"CLEANUP"事件。 ApplicationMasterLauncher维护了一个线程池， 从而能够尽快地处理这两种事件。

❑如果ApplicationMasterLauncher收到了"LAUNCH"类型的事件， 它会与对应的NodeManager通信， 要求它启动ApplicationMaster。 整个过程比较简单， 首先创建一个ContainerManagementProtocol协议的客户端， 然后向对应的NodeManager发起连接请求，接着将启动AM所需的各种信息， 包括启动命令、 JAR包、 环境变量等信息， 封装成一个StartContainerRequest对象， 然后通过RPC函数ContainerManagementProtocol\#startContainer发送给对应的NM。

❑如果ApplicationMasterLauncher收到了"CLEANUP"类型的事件， 它与对应的NodeManager通信， 要求它杀死ApplicationMaster。 整个过程与启动AM的过程类似。

LAUNCH和CLEANUP

## 3.1. 属性 ##

ApplicationMasterLauncher 继承于AbstractService . 所以 AbstractService 一些属性也是具备的,比如服务名称,服务的运行状态, 服务的开始时间等等属性. 当然它还有自己的属性.  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]  
ApplicationMasterLauncher 有四个比较重要的属性,  
RMContext用于存储RM的上下文信息.  
masterEvents是一个阻塞式的任务队列,用户缓冲要存储的任务时间.  
launcherPool是工作线程池, 默认数量为50 , 用于处理masterEvents队列汇总的事件任务.  
launcherHandlingThread是主线程, 无限从队列masterEvents中获取事件任务,交由launcherPool执行.

// ApplicationMasterLauncher 线程池
      private ThreadPoolExecutor launcherPool;
    
      // [ 主 ]工作线程, 用于监控masterEvents队列, 并将任务下发给launcherPool执行.
      private LauncherThread launcherHandlingThread;
    
      // 阻塞式队列
      private final BlockingQueue<Runnable> masterEvents  = new LinkedBlockingQueue<Runnable>();
    
      // RM的 context信息 : RMContextImpl
      protected final RMContext context;

## 3.2.初始化&启动 ##

ApplicationMasterLauncher的构建很简单,需要注意的是构造了一个主线程LauncherThread.  
先看一下构造方法:

public ApplicationMasterLauncher(RMContext context) { 
        super(ApplicationMasterLauncher.class.getName());
        this.context = context;
        this.launcherHandlingThread = new LauncherThread();
      }

然后看下初始化方法. 其实就是构建一个线程数为50的ThreadPoolExecutor线程池.

@Override
      protected void serviceInit(Configuration conf) throws Exception { 
    
        // 默认线程数量 : 50
        int threadCount = conf.getInt(
            YarnConfiguration.RM_AMLAUNCHER_THREAD_COUNT,
            YarnConfiguration.DEFAULT_RM_AMLAUNCHER_THREAD_COUNT);
        ThreadFactory tf = new ThreadFactoryBuilder()
            .setNameFormat("ApplicationMasterLauncher #%d")
            .build();
    
        // 构架线程池
        launcherPool = new ThreadPoolExecutor(threadCount, threadCount, 1,
            TimeUnit.HOURS, new LinkedBlockingQueue<Runnable>());
    
        // 构建线程工厂类
        launcherPool.setThreadFactory(tf);
    
        Configuration newConf = new YarnConfiguration(conf);
    
        // 最大重试次数: 10
        newConf.setInt(CommonConfigurationKeysPublic.
                IPC_CLIENT_CONNECT_MAX_RETRIES_ON_SOCKET_TIMEOUTS_KEY,
            conf.getInt(YarnConfiguration.RM_NODEMANAGER_CONNECT_RETRIES,
                YarnConfiguration.DEFAULT_RM_NODEMANAGER_CONNECT_RETRIES));
        setConfig(newConf);
        super.serviceInit(newConf);
      }

启动就更简单了,就是启动主线程就可以了.

@Override
      protected void serviceStart() throws Exception { 
        launcherHandlingThread.start();
        super.serviceStart();
      }

## 3.3.LauncherThread ##

LauncherThread 为ApplicationMasterLauncher 线程的主线程,声明周期是跟随RM一致,用于处理LAUNCH和CLEANUP 事件.

❑如果ApplicationMasterLauncher收到了"CLEANUP"类型的事件， 它与对应的NodeManager通信， 要求它杀死ApplicationMaster。 整个过程与启动AM的过程类似。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

在这里,主要是run方法, 从masterEvents阻塞式队列获取任务,然后交由工作线程池执行即可.

@Override
        public void run() { 
          while (!this.isInterrupted()) { 
            Runnable toLaunch;
            try { 
              // 从队列汇总获取任务.
              toLaunch = masterEvents.take();
              // 线程池用于处理 masterEvents 队列中的事件
              launcherPool.execute(toLaunch);
            } catch (InterruptedException e) { 
              LOG.warn(this.getClass().getName() + " interrupted. Returning.");
              return;
            }
          }
        }

## 3.3. 注册函数 ##

public synchronized void receivedPing(O ob) { 
        //only put for the registered objects
        if (running.containsKey(ob)) { 
          running.put(ob, clock.getTime());
        }
      }

## 3.4. 更新/移除注册\[心跳\]时间 ##

public synchronized void register(O ob) { 
        register(ob, clock.getTime());
      }
    
      public synchronized void register(O ob, long expireTime) { 
        running.put(ob, expireTime);
      }
    
      public synchronized void unregister(O ob) { 
        running.remove(ob);
      }

# 四 . AMLivelinessMonitor #

该服务周期性遍历所有应用程序的ApplicationMaster， 如果一个ApplicationMaster在一定时间（可通过参数yarn.am.liveness-monitor.expiry-interval-ms配置， 默认为10min） 内未汇报心跳信息， 则认为它死掉了， 它上面所有正在运行的Container将被置为运行失败（RM不会重新执行这些Container， 它只会通过心跳机制告诉对应的AM， 由AM决定是否重新执行。 如果需要， 则AM重新向RM申请资源） ； 如果AM运行失败， 则由RM重新为它申请资源， 以便能够重新分配到另外一个节点上（用户可在提交应用程序时通过函数ApplicationSubmissionContext\#setMaxAppAttempts设置ApplicationMaster重试次数， 如果未设置， 则采用全局参数yarn.resourcemanager.am.max-attempts设置的值， 默认是2） 执行。

## 4.1. AbstractLivelinessMonitor ##

一个简单的活跃度监视器，客户端可以用它注册，信任组件来监视活跃度，在过期时得到一个回调，然后最终取消注册。  
AMLivelinessMonitor 的父类是AbstractLivelinessMonitor, AbstractLivelinessMonitor是一个抽象类,它继承于AbstractService .

### 4.1.1. 属性 ###

在属性里面,需要重点关注的几个参数

1.  Thread checkerThread : 定时检测心跳的线程
2.  DEFAULT\_EXPIRE : 默认的过期时间为5min
3.  monitorInterval : 监控间隔, 为过期时间的 三分之一
4.  running : 监控程序的清单

// 定时检测心跳的线程
      private Thread checkerThread;
    
      // 标识是否停止.
      private volatile boolean stopped;
    
      // 默认过期时间 5min [ AMLivelinessMonitor 的过期时间为 10 min] 
      public static final int DEFAULT_EXPIRE = 5*60*1000;//5 mins
    
      // 过期时间
      private long expireInterval = DEFAULT_EXPIRE;
    
      // 监控间隔, 过期时间的 / 3
      private long monitorInterval = expireInterval / 3;
      
      // 在启动时 是否 重置Timer
      private volatile boolean resetTimerOnStart = true;
      
      // 时钟对象
      private final Clock clock;
    
      // 正在监控程序的清单   对象 -> 上一次心跳时间
      private Map<O, Long> running = new HashMap<O, Long>();

### 4.1.2. serviceStart 启动服务 ###

这里看到 使用PingChecker对象,构建了一个线程,用于监控任务.

@Override
      protected void serviceStart() throws Exception { 
        assert !stopped : "starting when already stopped";
        resetTimer();
        checkerThread = new Thread(new PingChecker());
        checkerThread.setName("Ping Checker");
        checkerThread.start();
        super.serviceStart();
      }

### 4.1.3. receivedPing ###

注册监控方法

public synchronized void receivedPing(O ob) {
        //only put for the registered objects
        if (running.containsKey(ob)) {
          running.put(ob, clock.getTime());
        }
      }

### 4.1.4. PingChecker 内部类 ###

PingChecker是AbstractLivelinessMonitor 的内部类.  
实现了Runnable方法, 主要是迭代running中记录的数据, 验证是否过期. 如果过期,触发回调函数expire.

@Override
        public void run() { 
          while (!stopped && !Thread.currentThread().isInterrupted()) { 
            synchronized (AbstractLivelinessMonitor.this) { 
    
              // 获取 running 中要监控对象的结合
              Iterator<Map.Entry<O, Long>> iterator = running.entrySet().iterator();
    
              // avoid calculating current time everytime in loop
              // 获取当前时间
              long currentTime = clock.getTime();
    
              while (iterator.hasNext()) { 
                // 获取监控对象
                Map.Entry<O, Long> entry = iterator.next();
                O key = entry.getKey();
                // 获取过期间隔
                long interval = getExpireInterval(key);
                // 判断是否过期,  当前时间 > 心跳时间+过期时间 ?
                if (currentTime > entry.getValue() + interval) { 
                  iterator.remove();
                  // 触发回调函数
                  expire(key);
                  LOG.info("Expired:" + entry.getKey().toString()
                      + " Timed out after " + interval / 1000 + " secs");
                }
              }
            }
            try { 
              Thread.sleep(monitorInterval);
            } catch (InterruptedException e) { 
              LOG.info(getName() + " thread interrupted");
              break;
            }
          }
        }

## 4.2.构造方法 ##

AMLivelinessMonitor 需要两个属性 Dispatcher d 和 Clock clock .

public AMLivelinessMonitor(Dispatcher d) { 
        super("AMLivelinessMonitor");
        this.dispatcher = d.getEventHandler();
      }
    
      public AMLivelinessMonitor(Dispatcher d, Clock clock) { 
        super("AMLivelinessMonitor", clock);
        this.dispatcher = d.getEventHandler();
      }

## 4.3. 初始化方法 ##

初始化是交由serviceInit方法完成的, 主要是设置默认过期时间10min和监控间隔, 10min的三分之一, 就是说大概三分30秒.监控一次.

public void serviceInit(Configuration conf) throws Exception { 
        super.serviceInit(conf);
        
        // 设置过期间隔: 10min
        int expireIntvl = conf.getInt(YarnConfiguration.RM_AM_EXPIRY_INTERVAL_MS,
                YarnConfiguration.DEFAULT_RM_AM_EXPIRY_INTERVAL_MS);
        setExpireInterval(expireIntvl);
        // 设置 监控间隔   , 过期间隔的三分之一
        setMonitorInterval(expireIntvl/3);
      }

## 4.4. 超时回调 ##

超时回调函数为expire . 入参为ApplicationAttemptId  
使用dispatcher进行处理.

@Override
      protected void expire(ApplicationAttemptId id) { 
        dispatcher.handle(
            new RMAppAttemptEvent(id, RMAppAttemptEventType.EXPIRE));
      }

# 五 . ApplicationMasterService #

ApplicationMasterService实现了RPC协议ApplicationMasterProtocol， 负责处理来自ApplicationMaster的请求，  
请求主要包括**注册、 心跳和清理**三种:  
**注册**是ApplicationMaster启动时发生的行为， 请求包中包含AM所在节点、 RPC端口号和trackingURL等信息；  
**心跳**是周期性行为， 包含请求资源的类型描述、 待释放的Container列表等，而AMS为之返回新分配的Container、 失败的Container等信息；  
**清理**是应用程序运行结束时发生的行为， ApplicationMaster向RM发送清理应用程序的请求， 以回收资源和清理各种内存空间。

ApplicationMasterLauncher启动AM后, AM做的第一件事是向RM注册， 这是通过RPC函数ApplicationMasterProtocol\#registerApplicationMaster实现的。AM运行过程中， 需要周期性地通过RPC函数ApplicationMasterProtocol\#allocate与RM通信，

这主要有以下三个作用：  
**❑请求资源；**  
**❑获取新分配的资源；**  
**❑形成周期性心跳， 告诉RM自己还活着。**

AM运行结束后， 需要通过RPC函数ApplicationMasterProtocol\#finishApplicationMaster告诉RM自己运行结束， 可以回收资源（AM所占用的Container） 和清理各种数据结果了（比如将该AM从AMLivelinessMonitor监控列表中删除） 。

## 5.1. 构造方法 ##

ApplicationMasterService的构造方法主要传入了AMLivelinessMonitor, YarnScheduler,RMContext和AMSProcessingChain属性.  
其中AMSProcessingChain通过责任链模式处理ApplicationMaster的注册 ,责任链上的processor的最后一个节点目前是DefaultAMSProcessor

public ApplicationMasterService(String name, RMContext rmContext,
          YarnScheduler scheduler) { 
        super(name);
        this.amLivelinessMonitor = rmContext.getAMLivelinessMonitor();
        this.rScheduler = scheduler;
        this.rmContext = rmContext;
        // AMSProcessingChain通过责任链模式处理ApplicationMaster的注册
        // 责任链上的processor的头节点目前是DefaultAMSProcessor
        this.amsProcessingChain = new AMSProcessingChain(new DefaultAMSProcessor());
      }

## 5.2. 初始化 ##

初始化方法构造了ApplicationMasterService发布服务的地址\[ 8030 \] 和构造AMSProcessingChain.

@Override
      protected void serviceInit(Configuration conf) throws Exception { 
        masterServiceAddress = conf.getSocketAddr(
            YarnConfiguration.RM_BIND_HOST,
            YarnConfiguration.RM_SCHEDULER_ADDRESS,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS,
            YarnConfiguration.DEFAULT_RM_SCHEDULER_PORT);
        initializeProcessingChain(conf);
      }

## 5.3. registerApplicationMaster ##

注册AM操作.

@Override
      public RegisterApplicationMasterResponse registerApplicationMaster(
          RegisterApplicationMasterRequest request) throws YarnException,
          IOException { 
    
        AMRMTokenIdentifier amrmTokenIdentifier =
            YarnServerSecurityUtils.authorizeRequest();
        ApplicationAttemptId applicationAttemptId =
            amrmTokenIdentifier.getApplicationAttemptId();
    
        // 获取ApplicationId
        ApplicationId appID = applicationAttemptId.getApplicationId();
        
        // 是否注册过 ? 获取注册响应对象  锁????
        // 因为在注册AM之前 会注册一个 app attempt
        AllocateResponseLock lock = responseMap.get(applicationAttemptId);
        if (lock == null) { 
          RMAuditLogger.logFailure(this.rmContext.getRMApps().get(appID).getUser(),
              AuditConstants.REGISTER_AM, "Application doesn't exist in cache "
                  + applicationAttemptId, "ApplicationMasterService",
              "Error in registering application master", appID,
              applicationAttemptId);
          throwApplicationDoesNotExistInCacheException(applicationAttemptId);
        }
    
        // 同一时间只能有一个线程注册
        // Allow only one thread in AM to do registerApp at a time.
        synchronized (lock) { 
    
          AllocateResponse lastResponse = lock.getAllocateResponse();
          if (hasApplicationMasterRegistered(applicationAttemptId)) { 
            // allow UAM re-register if work preservation is enabled
            ApplicationSubmissionContext appContext =
                rmContext.getRMApps().get(appID).getApplicationSubmissionContext();
            if (!(appContext.getUnmanagedAM()
                && appContext.getKeepContainersAcrossApplicationAttempts())) { 
              String message =
                  AMRMClientUtils.APP_ALREADY_REGISTERED_MESSAGE + appID;
              LOG.warn(message);
              RMAuditLogger.logFailure(
                  this.rmContext.getRMApps().get(appID).getUser(),
                  AuditConstants.REGISTER_AM, "", "ApplicationMasterService",
                  message, appID, applicationAttemptId);
              throw new InvalidApplicationMasterRequestException(message);
            }
          }
          // 更新心跳时间
          this.amLivelinessMonitor.receivedPing(applicationAttemptId);
    
          // 将 response id 设置为0，以标识应用程序主机是否注册了相应的 attemptid
          // Setting the response id to 0 to identify if the
          // application master is register for the respective attemptid
          lastResponse.setResponseId(0);
          // 更新lastResponse
          lock.setAllocateResponse(lastResponse);
    
          RegisterApplicationMasterResponse response =
              recordFactory.newRecordInstance(
                  RegisterApplicationMasterResponse.class);
    
          // 执行注册操作
          this.amsProcessingChain.registerApplicationMaster(amrmTokenIdentifier.getApplicationAttemptId(), request, response);
    
    
          return response;
        }
      }

## 5.5. finishApplicationMaster ##

@Override
      public FinishApplicationMasterResponse finishApplicationMaster(
          FinishApplicationMasterRequest request) throws YarnException,
          IOException { 
    
        // 获取 applicationAttemptId
        ApplicationAttemptId applicationAttemptId =
            YarnServerSecurityUtils.authorizeRequest().getApplicationAttemptId();
    
        // 获取 ApplicationId
        ApplicationId appId = applicationAttemptId.getApplicationId();
    
        // 获取RMApp
        RMApp rmApp =
            rmContext.getRMApps().get(applicationAttemptId.getApplicationId());
    
        // Remove collector address when app get finished.
        if (timelineServiceV2Enabled) { 
          ((RMAppImpl) rmApp).removeCollectorData(); } // checking whether the app exits in RMStateStore at first not to throw // ApplicationDoesNotExistInCacheException before and after // RM work-preserving restart. if (rmApp.isAppFinalStateStored()) { 
          LOG.info(rmApp.getApplicationId() + " unregistered successfully. ");
          return FinishApplicationMasterResponse.newInstance(true);
        }
    
        AllocateResponseLock lock = responseMap.get(applicationAttemptId);
        if (lock == null) { 
          throwApplicationDoesNotExistInCacheException(applicationAttemptId);
        }
    
        // Allow only one thread in AM to do finishApp at a time.
        synchronized (lock) { 
          if (!hasApplicationMasterRegistered(applicationAttemptId)) { 
            String message =
                "Application Master is trying to unregister before registering for: "
                    + appId;
            LOG.error(message);
            RMAuditLogger.logFailure(
                this.rmContext.getRMApps()
                    .get(appId).getUser(),
                AuditConstants.UNREGISTER_AM, "", "ApplicationMasterService",
                message, appId,
                applicationAttemptId);
            throw new ApplicationMasterNotRegisteredException(message);
          }
    
          FinishApplicationMasterResponse response =
              FinishApplicationMasterResponse.newInstance(false);
    
          // finishedAttemptCache 是否存在applicationAttemptId
          if (finishedAttemptCache.putIfAbsent(applicationAttemptId, true)
              == null) { 
            // 没有处理过,直接处理
            this.amsProcessingChain
                .finishApplicationMaster(applicationAttemptId, request, response);
          }
          // 处理监控心跳
          this.amLivelinessMonitor.receivedPing(applicationAttemptId);
          return response;
        }
      }

## 5.6. allocate ##

@Override
      public AllocateResponse allocate(AllocateRequest request)
          throws YarnException, IOException { 
    
        AMRMTokenIdentifier amrmTokenIdentifier =
            YarnServerSecurityUtils.authorizeRequest();
    
        ApplicationAttemptId appAttemptId =
            amrmTokenIdentifier.getApplicationAttemptId();
    
        // 更新心跳时间
        this.amLivelinessMonitor.receivedPing(appAttemptId);
    
        /* check if its in cache */
        AllocateResponseLock lock = responseMap.get(appAttemptId);
        if (lock == null) { 
          String message =
              "Application attempt " + appAttemptId
                  + " doesn't exist in ApplicationMasterService cache.";
          LOG.error(message);
          throw new ApplicationAttemptNotFoundException(message);
        }
        synchronized (lock) { 
          AllocateResponse lastResponse = lock.getAllocateResponse();
          if (!hasApplicationMasterRegistered(appAttemptId)) { 
            String message =
                "AM is not registered for known application attempt: "
                    + appAttemptId
                    + " or RM had restarted after AM registered. "
                    + " AM should re-register.";
            throw new ApplicationMasterNotRegisteredException(message);
          }
    
          // Normally request.getResponseId() == lastResponse.getResponseId()
          if (AMRMClientUtils.getNextResponseId(
              request.getResponseId()) == lastResponse.getResponseId()) { 
            // heartbeat one step old, simply return lastReponse
            return lastResponse;
          } else if (request.getResponseId() != lastResponse.getResponseId()) { 
            throw new InvalidApplicationMasterRequestException(AMRMClientUtils
                .assembleInvalidResponseIdExceptionMessage(appAttemptId,
                    lastResponse.getResponseId(), request.getResponseId()));
          }
    
          AllocateResponse response =
              recordFactory.newRecordInstance(AllocateResponse.class);
    
    	  // 关键点 ~~~~
          this.amsProcessingChain.allocate(
              amrmTokenIdentifier.getApplicationAttemptId(), request, response);
    
          // update AMRMToken if the token is rolled-up
          MasterKeyData nextMasterKey =
              this.rmContext.getAMRMTokenSecretManager().getNextMasterKeyData();
    
          if (nextMasterKey != null
              && nextMasterKey.getMasterKey().getKeyId() != amrmTokenIdentifier
              .getKeyId()) { 
    
            // 获取RM application
            RMApp app =  this.rmContext.getRMApps().get(appAttemptId.getApplicationId());
    
    
            RMAppAttempt appAttempt = app.getRMAppAttempt(appAttemptId);
    
            RMAppAttemptImpl appAttemptImpl = (RMAppAttemptImpl)appAttempt;
    
            Token<AMRMTokenIdentifier> amrmToken = appAttempt.getAMRMToken();
            if (nextMasterKey.getMasterKey().getKeyId() !=
                appAttemptImpl.getAMRMTokenKeyId()) { 
              LOG.info("The AMRMToken has been rolled-over. Send new AMRMToken back"
                  + " to application: " + appAttemptId.getApplicationId());
              amrmToken = rmContext.getAMRMTokenSecretManager()
                  .createAndGetAMRMToken(appAttemptId);
              appAttemptImpl.setAMRMToken(amrmToken);
            }
    
            response.setAMRMToken(org.apache.hadoop.yarn.api.records.Token
                .newInstance(amrmToken.getIdentifier(), amrmToken.getKind()
                    .toString(), amrmToken.getPassword(), amrmToken.getService()
                    .toString()));
          }
    
          /*
           * As we are updating the response inside the lock object so we don't
           * need to worry about unregister call occurring in between (which
           * removes the lock object).
           */
          response.setResponseId(
              AMRMClientUtils.getNextResponseId(lastResponse.getResponseId()));
          lock.setAllocateResponse(response);
          return response;
        }
      }

## 5.7. AMSProcessingChain ##

ApplicationMasterService中有一个属性AMSProcessingChain amsProcessingChain , AMSProcessingChain是以链表的方式进行存储,这样就可以通过责任链模式处理ApplicationMaster的注册, 责任链上的processor的最后一个是DefaultAMSProcessor . 这里面放了各种各样的约束程序,比如:DisabledPlacementProcessor,PlacementConstraintProcessor,SchedulerPlacementProcessor …

ApplicationMasterService中的registerApplicationMaster , finishApplicationMaster,allocate 方法都是交给DefaultAMSProcessor来实现的.

### 5.7.1. DefaultAMSProcessor\#registerApplicationMaster ###

ApplicationMasterService\#finishApplicationMaster 在注册application的时候调用.  
通过Dispatcher获取对应事件的Handler处理

@Override
      public void registerApplicationMaster(
          ApplicationAttemptId applicationAttemptId,
          RegisterApplicationMasterRequest request,
          RegisterApplicationMasterResponse response)
          throws IOException, YarnException { 
    
        // 获取 application
        RMApp app = getRmContext().getRMApps().get(
            applicationAttemptId.getApplicationId());
    
        //向RMAppAttemptImpl发送一个RMAppAttemptEventType.registered事件
        LOG.info("AM registration " + applicationAttemptId);
    
        // !!!! 在这里通过Dispatcher获取对应事件的Handler处理任务.!!!!!
        getRmContext().getDispatcher().getEventHandler()
            .handle(
                new RMAppAttemptRegistrationEvent(applicationAttemptId, request
                    .getHost(), request.getRpcPort(), request.getTrackingUrl()));
    
        // 日志审计
        RMAuditLogger.logSuccess(app.getUser(),
            RMAuditLogger.AuditConstants.REGISTER_AM,
            "ApplicationMasterService", app.getApplicationId(),
            applicationAttemptId);
    
        // 返回资源的最大容量
        response.setMaximumResourceCapability(getScheduler()
            .getMaximumResourceCapability(app.getQueue()));
    
        // 设置权限
        response.setApplicationACLs(app.getRMAppAttempt(applicationAttemptId)
            .getSubmissionContext().getAMContainerSpec().getApplicationACLs());
    
        // 设置队列
        response.setQueue(app.getQueue());
        if (UserGroupInformation.isSecurityEnabled()) { 
          LOG.info("Setting client token master key");
          response.setClientToAMTokenMasterKey(java.nio.ByteBuffer.wrap(
              getRmContext().getClientToAMTokenSecretManager()
              .getMasterKey(applicationAttemptId).getEncoded()));
        }
    
        // For work-preserving AM restart, retrieve previous attempts' containers // and corresponding NM tokens. if (app.getApplicationSubmissionContext() .getKeepContainersAcrossApplicationAttempts()) { List<Container> transferredContainers = getScheduler() .getTransferredContainers(applicationAttemptId); if (!transferredContainers.isEmpty()) { response.setContainersFromPreviousAttempts(transferredContainers); // Clear the node set remembered by the secret manager. Necessary // for UAM restart because we use the same attemptId. rmContext.getNMTokenSecretManager() .clearNodeSetForAttempt(applicationAttemptId); List<NMToken> nmTokens = new ArrayList<NMToken>(); for (Container container : transferredContainers) { try { NMToken token = getRmContext().getNMTokenSecretManager() .createAndGetNMToken(app.getUser(), applicationAttemptId, container); if (null != token) { nmTokens.add(token); } } catch (IllegalArgumentException e) { // if it's a DNS issue, throw UnknowHostException directly and
                // that
                // will be automatically retried by RMProxy in RPC layer.
                if (e.getCause() instanceof UnknownHostException) { 
                  throw (UnknownHostException) e.getCause();
                }
              }
            }
    
            response.setNMTokensFromPreviousAttempts(nmTokens);
            LOG.info("Application " + app.getApplicationId() + " retrieved "
                + transferredContainers.size() + " containers from previous"
                + " attempts and " + nmTokens.size() + " NM tokens.");
          }
        }
    
        // 设置资源类型
        response.setSchedulerResourceTypes(getScheduler()
            .getSchedulingResourceTypes());
    
        // 设置资源类型
        response.setResourceTypes(ResourceUtils.getResourcesTypeInfo());
        if (getRmContext().getYarnConfiguration().getBoolean(
            YarnConfiguration.RM_RESOURCE_PROFILES_ENABLED,
            YarnConfiguration.DEFAULT_RM_RESOURCE_PROFILES_ENABLED)) { 
          response.setResourceProfiles(
              resourceProfilesManager.getResourceProfiles());
        }
      }

### 5.7.2. DefaultAMSProcessor\#finishApplicationMaster ###

ApplicationMasterService\#finishApplicationMaster 在完成application的时候调用.  
通过Dispatcher获取对应事件的Handler处理

@Override
      public void finishApplicationMaster(
          ApplicationAttemptId applicationAttemptId,
          FinishApplicationMasterRequest request,
          FinishApplicationMasterResponse response) { 
    
        // 获取application
        RMApp app =
            getRmContext().getRMApps().get(applicationAttemptId.getApplicationId());
        // For UnmanagedAMs, return true so they don't retry
        response.setIsUnregistered(
                app.getApplicationSubmissionContext().getUnmanagedAM());
    
        // 通过Dispatcher处理对应的时间
        getRmContext().getDispatcher().getEventHandler().handle(
            new RMAppAttemptUnregistrationEvent(applicationAttemptId, request
                .getTrackingUrl(), request.getFinalApplicationStatus(), request
                .getDiagnostics()));
      }

DisabledPlacementProcessor  
PlacementConstraintProcessor  
SchedulerPlacementProcessor

### 5.7.3. DefaultAMSProcessor\#allocate ###

更新资源/心跳.

@Override
      public void allocate(ApplicationAttemptId appAttemptId,
          AllocateRequest request, AllocateResponse response) throws YarnException { 
        // 处理 请求
        handleProgress(appAttemptId, request);
    
        // 获取资源请求
        List<ResourceRequest> ask = request.getAskList();
    
        // 获取需要释放的 ContainerId
        List<ContainerId> release = request.getReleaseList();
    
        // 获取请求的黑名单
        ResourceBlacklistRequest blacklistRequest =  request.getResourceBlacklistRequest();
        List<String> blacklistAdditions =
            (blacklistRequest != null) ?
                blacklistRequest.getBlacklistAdditions() : Collections.emptyList();
    
        // 获取移除黑名单的数据
        List<String> blacklistRemovals =
            (blacklistRequest != null) ?
                blacklistRequest.getBlacklistRemovals() : Collections.emptyList();
    
    
        // 获取 RM application
        RMApp app =
            getRmContext().getRMApps().get(appAttemptId.getApplicationId());
    
        // set label expression for Resource Requests if resourceName=ANY
        // 根据资源请求 设置 label表达式
        ApplicationSubmissionContext asc = app.getApplicationSubmissionContext();
        for (ResourceRequest req : ask) { 
          if (null == req.getNodeLabelExpression()
              && ResourceRequest.ANY.equals(req.getResourceName())) { 
            req.setNodeLabelExpression(asc.getNodeLabelExpression());
          }
        }
    
        // 获取队列中的最大资源能力
        Resource maximumCapacity =
            getScheduler().getMaximumResourceCapability(app.getQueue());
    
        // 健全性检查
        // sanity check
        try { 
          RMServerUtils.normalizeAndValidateRequests(ask,
              maximumCapacity, app.getQueue(),
              getScheduler(), getRmContext(), nodelabelsEnabled);
        } catch (InvalidResourceRequestException e) { 
          RMAppAttempt rmAppAttempt = app.getRMAppAttempt(appAttemptId);
          handleInvalidResourceException(e, rmAppAttempt);
        }
    
        try { 
          // 验证请求
          RMServerUtils.validateBlacklistRequest(blacklistRequest);
        }  catch (InvalidResourceBlacklistRequestException e) { 
          LOG.warn("Invalid blacklist request by application " + appAttemptId, e);
          throw e;
        }
        // 在工作保持AM重启的情况下，AM有可能从先前的尝试中释放容器。
        // In the case of work-preserving AM restart, it's possible for the
        // AM to release containers from the earlier attempt.
        if (!app.getApplicationSubmissionContext()
            .getKeepContainersAcrossApplicationAttempts()) { 
          try { 
            RMServerUtils.validateContainerReleaseRequest(release, appAttemptId);
          } catch (InvalidContainerReleaseException e) { 
            LOG.warn("Invalid container release by application " + appAttemptId,
                e);
            throw e;
          }
        }
    
        // 将更新资源请求拆分为增加和减少。
        // 这里没有抛出异常。所有更新错误都将聚合并返回给AM。
        // Split Update Resource Requests into increase and decrease.
        // No Exceptions are thrown here. All update errors are aggregated
        // and returned to the AM.
        List<UpdateContainerError> updateErrors = new ArrayList<>();
        ContainerUpdates containerUpdateRequests =
            RMServerUtils.validateAndSplitUpdateResourceRequests(
                getRmContext(), request, maximumCapacity, updateErrors);
    
        // 向appAttempt 发送新的requests
        // Send new requests to appAttempt.
        Allocation allocation;
        RMAppAttemptState state =
            app.getRMAppAttempt(appAttemptId).getAppAttemptState();
    
        if (state.equals(RMAppAttemptState.FINAL_SAVING) ||
            state.equals(RMAppAttemptState.FINISHING) ||
            app.isAppFinalStateStored()) { 
          // 如果任务正处于保存中或者保存完成 , 不做操作,修改allocation状态为EMPTY_ALLOCATION
          LOG.warn(appAttemptId + " is in " + state +
              " state, ignore container allocate request.");
          allocation = EMPTY_ALLOCATION;
        } else { 
          try { 
    
            // 重点!!!  获取 YarnScheduler 处理 allocate
            allocation = getScheduler().allocate(appAttemptId, ask,
                request.getSchedulingRequests(), release,
                blacklistAdditions, blacklistRemovals, containerUpdateRequests);
          } catch (SchedulerInvalidResoureRequestException e) { 
            LOG.warn("Exceptions caught when scheduler handling requests");
            throw new YarnException(e);
          }
        }
    
        if (!blacklistAdditions.isEmpty() || !blacklistRemovals.isEmpty()) { 
          LOG.info("blacklist are updated in Scheduler." +
              "blacklistAdditions: " + blacklistAdditions + ", " +
              "blacklistRemovals: " + blacklistRemovals);
        }
        // 获取RMAppAttempt
        RMAppAttempt appAttempt = app.getRMAppAttempt(appAttemptId);
    
        if (allocation.getNMTokens() != null &&
            !allocation.getNMTokens().isEmpty()) { 
          response.setNMTokens(allocation.getNMTokens());
        }
    
        // 通知AM的container更新错误信息
        // Notify the AM of container update errors
        ApplicationMasterServiceUtils.addToUpdateContainerErrors(
            response, updateErrors);
    
        // 使用节点状态更改的增量更新响应
        // update the response with the deltas of node status changes
        handleNodeUpdates(app, response);
    
        // 添加新分配的container...
        ApplicationMasterServiceUtils.addToAllocatedContainers(
            response, allocation.getContainers());
    
        // 设置已完成Container状态
        response.setCompletedContainersStatuses(appAttempt
            .pullJustFinishedContainers());
        // 设置可用资源
        response.setAvailableResources(allocation.getResourceLimit());
    
        // 设置Container变更
        addToContainerUpdates(response, allocation,
            ((AbstractYarnScheduler)getScheduler())
                .getApplicationAttempt(appAttemptId).pullUpdateContainerErrors());
    
        // 设置集群节点数量
        response.setNumClusterNodes(getScheduler().getNumClusterNodes());
    
        // add collector address for this application
        if (timelineServiceV2Enabled) { 
          CollectorInfo collectorInfo = app.getCollectorInfo();
          if (collectorInfo != null) { 
            response.setCollectorInfo(collectorInfo);
          }
        }
    
        // 向allocateResponse消息添加抢占（如果有）
        // add preemption to the allocateResponse message (if any)
        response.setPreemptionMessage(generatePreemptionMessage(allocation));
    
        //  设置application 优先级
        // Set application priority
        response.setApplicationPriority(app
            .getApplicationPriority());
        
        response.setContainersFromPreviousAttempts(
            allocation.getPreviousAttemptContainers());
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221120/0e3b5a1445034f5ab1051648fe485ad5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221120/385b5eb154ae490aaf1701371b0cb6ff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221120/fe73fd957c764413b0784c274fb5aba9.png