import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public synchronized void transitionToStandby(
    HAServiceProtocol.StateChangeRequestInfo reqInfo) throws IOException {
  // call refreshAdminAcls before HA state transition
  // for the case that adminAcls have been updated in previous active RM
  try {
    refreshAdminAcls(false);
  } catch (YarnException ex) {
    throw new ServiceFailedException("Can not execute refreshAdminAcls", ex);
  }
  UserGroupInformation user = checkAccess("transitionToStandby");
  checkHaStateChange(reqInfo);
  try {
    rm.transitionToStandby(true);
    RMAuditLogger.logSuccess(user.getShortUserName(),
        "transitionToStandby", "RMHAProtocolService");
  } catch (Exception e) {
    RMAuditLogger.logFailure(user.getShortUserName(), "transitionToStandby",
        "", "RMHAProtocolService",
        "Exception transitioning to standby");
    throw new ServiceFailedException(
        "Error when transitioning to Standby mode", e);
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
private void refreshAll() throws ServiceFailedException {
  try {
    refreshQueues(RefreshQueuesRequest.newInstance());
    refreshNodes(RefreshNodesRequest.newInstance());
    refreshSuperUserGroupsConfiguration(
        RefreshSuperUserGroupsConfigurationRequest.newInstance());
    refreshUserToGroupsMappings(
        RefreshUserToGroupsMappingsRequest.newInstance());
    if (getConfig().getBoolean(
        CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
        false)) {
      refreshServiceAcls(RefreshServiceAclsRequest.newInstance());
    }
  } catch (Exception ex) {
    throw new ServiceFailedException(ex.getMessage());
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
protected AdminService createAdminService() {
  return new AdminService(MockRMWithElector.this, getRMContext()) {
    @Override
    protected EmbeddedElectorService createEmbeddedElectorService() {
      return new EmbeddedElectorService(getRMContext()) {
        @Override
        public void becomeActive() throws
            ServiceFailedException {
          try {
            callbackCalled.set(true);
            LOG.info("Callback called. Sleeping now");
            Thread.sleep(delayMs);
            LOG.info("Sleep done");
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          super.becomeActive();
        }
      };
    }
  };
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
/**
 * Cancel any checkpoint that's currently being made,
 * and prevent any new checkpoints from starting for the next
 * minute or so.
 */
public void cancelAndPreventCheckpoints(String msg) throws ServiceFailedException {
  synchronized (cancelLock) {
    // The checkpointer thread takes this lock and checks if checkpointing is
    // postponed. 
    thread.preventCheckpointsFor(PREVENT_AFTER_CANCEL_MS);

    // Before beginning a checkpoint, the checkpointer thread
    // takes this lock, and creates a canceler object.
    // If the canceler is non-null, then a checkpoint is in
    // progress and we need to cancel it. If it's null, then
    // the operation has not started, meaning that the above
    // time-based prevention will take effect.
    if (canceler != null) {
      canceler.cancel(msg);
    }
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
private void assertCanStartHaNameNodes(String pathSuffix)
    throws ServiceFailedException, IOException, URISyntaxException,
    InterruptedException {
  // Now should be able to start both NNs. Pass "false" here so that we don't
  // try to waitActive on all NNs, since the second NN doesn't exist yet.
  cluster.restartNameNode(0, false);
  cluster.restartNameNode(1, true);
  
  // Make sure HA is working.
  cluster.getNameNode(0).getRpcServer().transitionToActive(
      new StateChangeRequestInfo(RequestSource.REQUEST_BY_USER));
  FileSystem fs = null;
  try {
    Path newPath = new Path(TEST_PATH, pathSuffix);
    fs = HATestUtil.configureFailoverFs(cluster, conf);
    assertTrue(fs.mkdirs(newPath));
    HATestUtil.waitForStandbyToCatchUp(cluster.getNameNode(0),
        cluster.getNameNode(1));
    assertTrue(NameNodeAdapter.getFileInfo(cluster.getNameNode(1),
        newPath.toString(), false).isDir());
  } finally {
    if (fs != null) {
      fs.close();
    }
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
private void refreshAll() throws ServiceFailedException {
  try {
    refreshQueues(RefreshQueuesRequest.newInstance());
    refreshNodes(RefreshNodesRequest.newInstance(DecommissionType.NORMAL));
    refreshSuperUserGroupsConfiguration(
        RefreshSuperUserGroupsConfigurationRequest.newInstance());
    refreshUserToGroupsMappings(
        RefreshUserToGroupsMappingsRequest.newInstance());
    if (getConfig().getBoolean(
        CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
        false)) {
      refreshServiceAcls(RefreshServiceAclsRequest.newInstance());
    }
    refreshClusterMaxPriority(RefreshClusterMaxPriorityRequest.newInstance());
  } catch (Exception ex) {
    throw new ServiceFailedException(ex.getMessage());
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
protected AdminService createAdminService() {
  return new AdminService(MockRMWithElector.this, getRMContext()) {
    @Override
    protected EmbeddedElectorService createEmbeddedElectorService() {
      return new EmbeddedElectorService(getRMContext()) {
        @Override
        public void becomeActive() throws
            ServiceFailedException {
          try {
            callbackCalled.set(true);
            TestRMEmbeddedElector.LOG.info("Callback called. Sleeping now");
            Thread.sleep(delayMs);
            TestRMEmbeddedElector.LOG.info("Sleep done");
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          super.becomeActive();
        }
      };
    }
  };
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
synchronized HAServiceStatus getServiceStatus()
    throws ServiceFailedException, AccessControlException {
  namesystem.checkSuperuserPrivilege();
  if (!haEnabled) {
    throw new ServiceFailedException("HA for namenode is not enabled");
  }
  if (state == null) {
    return new HAServiceStatus(HAServiceState.INITIALIZING);
  }
  HAServiceState retState = state.getServiceState();
  HAServiceStatus ret = new HAServiceStatus(retState);
  if (retState == HAServiceState.STANDBY) {
    if (namesystem.isInSafeMode()) {
      ret.setNotReadyToBecomeActive("The NameNode is in safemode. " +
          namesystem.getSafeModeTip());
    } else {
      ret.setReadyToBecomeActive();
    }
  } else if (retState == HAServiceState.ACTIVE) {
    ret.setReadyToBecomeActive();
  } else {
    ret.setNotReadyToBecomeActive("State is " + state);
  }
  return ret;
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
/**
* Stop all NameNode threads and wait for all to finish.
*/
 public void stop() {
     synchronized (this) {
         if (stopRequested) {
             return;
         }
         stopRequested = true;
     }
     try {
         exitActiveServices();
     } catch (ServiceFailedException e) {
         LOG.warn("Encountered exception while exiting state ", e);
     } finally {
         stopCommonServices();
         if (metrics != null) {
             metrics.shutdown();
         }
         if (namesystem != null) {
             namesystem.shutdown();
         }
     }
 }
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public void transitionToStandby(StateChangeRequestInfo reqInfo)
    throws ServiceFailedException, AccessControlException, IOException {
  if (haState == HAServiceState.STANDBY) {
    LOG.info("Already in standby state.");
    return;
  }
  LOG.info("Transitioning to standby");
  try {
    if (sysDirMonitorExecutor != null) {
      sysDirMonitorExecutor.shutdownNow();
    }
    jtRunner.stopJobTracker();
    httpRedirector.start();
  } catch (Throwable t) {
    doImmediateShutdown(t);
  }
  sysDirMonitorExecutor = null;
  currentSysDir = null;
  haState = HAServiceState.STANDBY;
  LOG.info("Transitioned to standby");
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
private void refreshAll() throws ServiceFailedException {
  try {
    refreshQueues(RefreshQueuesRequest.newInstance());
    refreshNodes(RefreshNodesRequest.newInstance(DecommissionType.NORMAL));
    refreshSuperUserGroupsConfiguration(
        RefreshSuperUserGroupsConfigurationRequest.newInstance());
    refreshUserToGroupsMappings(
        RefreshUserToGroupsMappingsRequest.newInstance());
    if (getConfig().getBoolean(
        CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION,
        false)) {
      refreshServiceAcls(RefreshServiceAclsRequest.newInstance());
    }
    refreshClusterMaxPriority(RefreshClusterMaxPriorityRequest.newInstance());
  } catch (Exception ex) {
    throw new ServiceFailedException("RefreshAll operation failed", ex);
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
protected EmbeddedElector createEmbeddedElector() {
  return new ActiveStandbyElectorBasedElectorService(getRMContext()) {
    @Override
    public void becomeActive() throws
        ServiceFailedException {
      try {
        callbackCalled.set(true);
        TestRMEmbeddedElector.LOG.info("Callback called. Sleeping now");
        Thread.sleep(delayMs);
        TestRMEmbeddedElector.LOG.info("Sleep done");
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
      super.becomeActive();
    }
  };
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public synchronized void transitionToStandby(
    HAServiceProtocol.StateChangeRequestInfo reqInfo) throws IOException {
  // call refreshAdminAcls before HA state transition
  // for the case that adminAcls have been updated in previous active RM
  try {
    refreshAdminAcls(false);
  } catch (YarnException ex) {
    throw new ServiceFailedException("Can not execute refreshAdminAcls", ex);
  }
  UserGroupInformation user = checkAccess("transitionToStandby");
  checkHaStateChange(reqInfo);
  try {
    rm.transitionToStandby(true);
    RMAuditLogger.logSuccess(user.getShortUserName(),
        "transitionToStandby", "RM");
  } catch (Exception e) {
    RMAuditLogger.logFailure(user.getShortUserName(), "transitionToStandby",
        "", "RM",
        "Exception transitioning to standby");
    throw new ServiceFailedException(
        "Error when transitioning to Standby mode", e);
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public void becomeActive() throws ServiceFailedException {
  try {
    rmContext.getRMAdminService().transitionToActive(req);
  } catch (Exception e) {
    throw new ServiceFailedException("RM could not transition to Active", e);
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public synchronized void transitionToActive(
    HAServiceProtocol.StateChangeRequestInfo reqInfo) throws IOException {
  // call refreshAdminAcls before HA state transition
  // for the case that adminAcls have been updated in previous active RM
  try {
    refreshAdminAcls(false);
  } catch (YarnException ex) {
    throw new ServiceFailedException("Can not execute refreshAdminAcls", ex);
  }

  UserGroupInformation user = checkAccess("transitionToActive");
  checkHaStateChange(reqInfo);
  try {
    rm.transitionToActive();
    // call all refresh*s for active RM to get the updated configurations.
    refreshAll();
    RMAuditLogger.logSuccess(user.getShortUserName(),
        "transitionToActive", "RMHAProtocolService");
  } catch (Exception e) {
    RMAuditLogger.logFailure(user.getShortUserName(), "transitionToActive",
        "", "RMHAProtocolService",
        "Exception transitioning to active");
    throw new ServiceFailedException(
        "Error when transitioning to Active mode", e);
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public void setState(HAContext context, HAState s) throws ServiceFailedException {
  if (s == NameNode.STANDBY_STATE) {
    setStateInternal(context, s);
    return;
  }
  super.setState(context, s);
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public void enterState(HAContext context) throws ServiceFailedException {
  try {
    context.startActiveServices();
  } catch (IOException e) {
    throw new ServiceFailedException("Failed to start active services", e);
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
@Override
public void exitState(HAContext context) throws ServiceFailedException {
  try {
    context.stopActiveServices();
  } catch (IOException e) {
    throw new ServiceFailedException("Failed to stop active services", e);
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
/**
 * Internal method to transition the state of a given namenode to a new state.
 * @param nn Namenode
 * @param s new state
 * @throws ServiceFailedException on failure to transition to new state.
 */
protected final void setStateInternal(final HAContext context, final HAState s)
    throws ServiceFailedException {
  prepareToExitState(context);
  s.prepareToEnterState(context);
  context.writeLock();
  try {
    exitState(context);
    context.setState(s);
    s.enterState(context);
    s.updateLastHATransitionTime();
  } finally {
    context.writeUnlock();
  }
}
 

import org.apache.hadoop.ha.ServiceFailedException; //导入依赖的package包/类
/**
 * Move from the existing state to a new state
 * @param context HA context
 * @param s new state
 * @throws ServiceFailedException on failure to transition to new state.
 */
public void setState(HAContext context, HAState s) throws ServiceFailedException {
  if (this == s) { // Aleady in the new state
    return;
  }
  throw new ServiceFailedException("Transtion from state " + this + " to "
      + s + " is not allowed.");
}