ZooKeeper节点数据量限制引起的Hadoop YARN ResourceManager崩溃原因分析（三）

2020-04-28 10:05:54 INFO  org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore:768 - Application update attemptState data size for /rmstore/ZKRMStateRoot/RMAppRoot/application_1587969707206_16259/appattempt_1587969707206_16259_000001 is 20266528. Exceed the maximum allowed 3145728 size. ApplicationAttemptState info: ApplicationAttemptState{attemptId=appattempt_1587969707206_16259_000001, diagnostics='User class threw exception: java.lang.RuntimeException: org.apache.spark.SparkException: Job aborted due to stage failure: Task 15 in stage 2.0 failed 4 times, most recent failure: Lost task 15.3 in stage 2.0 (TID 4224, datanode44.bi): java.util.concurrent.ExecutionException: java.lang.Exception: failed to compile: org.codehaus.janino.JaninoRuntimeException: Code of method "(Lorg/apache/spark/sql/catalyst/expressions/GeneratedClass$SpecificUnsafeProjection;Lorg/apache/spark/sql/catalyst/InternalRow;)V" of class "org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection" grows beyond 64 KB
/* 001 */ public java.lang.Object generate(Object[] references) {
/* 002 */   return new SpecificUnsafeProjection(references);
/* 003 */ }
/* 004 */
/* 005 */ class SpecificUnsafeProjection extends org.apache.spark.sql.catalyst.expressions.UnsafeProjection {
/* 006 */
/* 007 */   private Object[] references;
/* 008 */   private scala.collection.immutable.Set hset;
/* 009 */   private boolean hasNull;
/* 010 */   private UnsafeRow result;
/* 011 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder holder;
/* 012 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter rowWriter;

public class ZKRMStateStore extends RMStateStore{
    
    private int zknodeLimit; // 保存ZNode节点数据量限制
    
     @Override
    public synchronized void initInternal(Configuration conf) throws Exception {
        // 其余部分省略
        // 获取yarn-site.xml中yarn.resourcemanager.zk-max-znode-size.bytes的值
        zknodeLimit = conf.getInt(YarnConfiguration.RM_ZK_ZNODE_SIZE_LIMIT_BYTES,
                YarnConfiguration.DEFAULT_RM_ZK_ZNODE_SIZE_LIMIT_BYTES);
    }
    
    @Override
    public synchronized void updateApplicationAttemptStateInternal(
            ApplicationAttemptId appAttemptId,
            ApplicationAttemptStateData attemptStateDataPB)
            throws Exception {
        String appIdStr = appAttemptId.getApplicationId().toString();
        String appAttemptIdStr = appAttemptId.toString();
        String appDirPath = getNodePath(rmAppRoot, appIdStr);
        String nodeUpdatePath = getNodePath(appDirPath, appAttemptIdStr);
        if (LOG.isDebugEnabled()) {
            LOG.debug("Storing final state info for attempt: " + appAttemptIdStr
                    + " at: " + nodeUpdatePath);
        }
        byte[] attemptStateData = attemptStateDataPB.getProto().toByteArray();

        ApplicationAttemptState attemptState = getApplicationAttemptState(appAttemptId, attemptStateDataPB);
        // 判断要写入的任务尝试数据信息是否超过zknodeLimit变量的值，如果没有，就执行任务尝试数据更新操作。否则，只打印info信息，不执行任务尝试数据更新操作
        if (attemptStateData.length <= zknodeLimit) {
            if (existsWithRetries(nodeUpdatePath, true) != null) {
                setDataWithRetries(nodeUpdatePath, attemptStateData, -1);
            } else {
                createWithRetries(nodeUpdatePath, attemptStateData, zkAcl,
                        CreateMode.PERSISTENT);
                LOG.debug(appAttemptId + " znode didn't exist. Created a new znode to"
                        + " update the application attempt state.");
            }
            LOG.info("Application update attemptState data size for " + nodeUpdatePath + " is "
                    + attemptStateData.length + ". The maximum allowed " + zknodeLimit + " size. ApplicationAttemptState info: " + attemptState.toString() + ". AppAttemptTokens length:" + attemptStateDataPB.getAppAttemptTokens().array().length
                    + ". See yarn.resourcemanager.zk-max-znode-size.bytes.");
        } else {
            LOG.info("Application update attemptState data size for " + nodeUpdatePath + " is "
                    + attemptStateData.length + ". Exceed the maximum allowed " + zknodeLimit + " size. ApplicationAttemptState info: " + attemptState.toString() + ". AppAttemptTokens length:" + attemptStateDataPB.getAppAttemptTokens().array().length
                    + ". See yarn.resourcemanager.zk-max-znode-size.bytes.");
        }
    }
}

 /**
   * State of an application application
   * 任务状态信息类
   */
  public static class ApplicationState {
    final ApplicationSubmissionContext context; // 任务描述信息content
    final long submitTime; // 任务提交时间
    final long startTime; // 任务开始时间
    final String user; // 任务提交人
    Map<ApplicationAttemptId, ApplicationAttemptState> attempts =
                  new HashMap<ApplicationAttemptId, ApplicationAttemptState>(); // 任务重试信息
    // fields set when application completes.
    RMAppState state; // 任务运行状态
    String diagnostics; // 任务异常诊断信息
    long finishTime; // 任务完成时间
    
    // 省略其他代码
  }

 /**
   * State of an application attempt
   * 任务尝试状态信息类
   */
  public static class ApplicationAttemptState {
    final ApplicationAttemptId attemptId; // 任务尝试ID
    final Container masterContainer; // 所在container的信息
    final Credentials appAttemptCredentials; // 安全token
    long startTime = 0; // 开始时间
    long finishTime = 0; // 结束时间
    // fields set when attempt completes
    RMAppAttemptState state; // 运行状态
    String finalTrackingUrl = "N/A"; // 任务运行日志查看地址
    String diagnostics; // 任务异常诊断信息
    int exitStatus = ContainerExitStatus.INVALID; // 任务退出状态
    FinalApplicationStatus amUnregisteredFinalStatus; // 任务最终状态
    long memorySeconds; // 任务消耗的内存总资源
    long vcoreSeconds; // 任务消耗的CPU总资源
    
    // 省略其他代码
  }

 /**
   * 安全令牌信息
   */
  public static class RMDTSecretManagerState {
    // DTIdentifier -> renewDate
    Map<RMDelegationTokenIdentifier, Long> delegationTokenState =
        new HashMap<RMDelegationTokenIdentifier, Long>(); // 授权令牌状态
    Set<DelegationKey> masterKeyState =
        new HashSet<DelegationKey>(); // master key状态
    int dtSequenceNumber = 0; // 序列号
    
    // 省略其他代码
 }

 private synchronized void loadRMAppState(RMState rmState) throws Exception {
        // 当/rmstore/ZKRMStateRoot/RMAppRoot/节点及其子节点被删除或创建时，watch被触发
        List<String> childNodes = getChildrenWithRetries(rmAppRoot, true);
        for (String childNodeName : childNodes) {
            String childNodePath = getNodePath(rmAppRoot, childNodeName);
            // 获取任务节点数据并注册watch，该watch当任务节点被删除或数据被更新时触发
            byte[] childData = getDataWithRetries(childNodePath, true);
            if (childNodeName.startsWith(ApplicationId.appIdStrPrefix)) {
                // application
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Loading application from znode: " + childNodeName);
                }
                ApplicationId appId = ConverterUtils.toApplicationId(childNodeName);
                ApplicationStateDataPBImpl appStateData =
                        new ApplicationStateDataPBImpl(
                                ApplicationStateDataProto.parseFrom(childData));
                // 获取任务数据
                ApplicationState appState =
                        new ApplicationState(appStateData.getSubmitTime(),
                                appStateData.getStartTime(),
                                appStateData.getApplicationSubmissionContext(),
                                appStateData.getUser(),
                                appStateData.getState(),
                                appStateData.getDiagnostics(), appStateData.getFinishTime());
                if (!appId.equals(appState.context.getApplicationId())) {
                    throw new YarnRuntimeException("The child node name is different " +
                            "from the application id");
                }
                rmState.appState.put(appId, appState);
                // 获取任务重试数据
                loadApplicationAttemptState(appState, appId);
            } else {
                LOG.info("Unknown child node with name: " + childNodeName);
            }
        }
    }

 /**
   * 该方法通知RM存储操作失败，参数是引起操作失败的异常信息
   * This method is called to notify the ResourceManager that the store
   * operation has failed.
   * @param failureCause the exception due to which the operation failed
   */
  protected void notifyStoreOperationFailed(Exception failureCause) {
    LOG.error("State store operation failed ", failureCause);
    // 如果开启了HA，则执行故障转移操作
    if (HAUtil.isHAEnabled(getConfig())) {
      LOG.warn("State-store fenced ! Transitioning RM to standby");
      Thread standByTransitionThread =
          new Thread(new StandByTransitionThread());
      standByTransitionThread.setName("StandByTransitionThread Handler");
      standByTransitionThread.start();
    } else if (YarnConfiguration.shouldRMFailFast(getConfig())) { // 如果没有开启HA，则判断有没有开启快速失败
      LOG.fatal("Fail RM now due to state-store error!");
      rmDispatcher.getEventHandler().handle(
          new RMFatalEvent(RMFatalEventType.STATE_STORE_OP_FAILED,
              failureCause));
    } else { // 否则，打印跳过存储异常警告信息
      LOG.warn("Skip the state-store error.");
    }
  }