@Override
  public void start() throws Exception {
    LOG.info("Process starting.");
    mRunning = true;
    mJournalSystem.start();
    startMasterComponents(false);
    mServices.forEach(SimpleService::start);

    // Perform the initial catchup before joining leader election,
    // to avoid potential delay if this master is selected as leader
    if (Configuration.getBoolean(PropertyKey.MASTER_JOURNAL_CATCHUP_PROTECT_ENABLED)) {
      LOG.info("Waiting for journals to catch up.");
      mJournalSystem.waitForCatchup();
    }

    LOG.info("Starting leader selector.");
    mLeaderSelector.start(getRpcAddress());

    while (!Thread.interrupted()) {
      if (!mRunning) {
        LOG.info("master process is not running. Breaking out");
        break;
      }
      if (Configuration.getBoolean(PropertyKey.MASTER_JOURNAL_CATCHUP_PROTECT_ENABLED)) {
        LOG.info("Waiting for journals to catch up.");
        mJournalSystem.waitForCatchup();
      }

      LOG.info("Started in stand-by mode.");
      mLeaderSelector.waitForState(NodeState.PRIMARY);
      mLastGainPrimacyTime = CommonUtils.getCurrentMs();
      if (!mRunning) {
        break;
      }
      try {
        if (!promote()) {
          continue;
        }
        mServices.forEach(SimpleService::promote);
        LOG.info("Primary started");
      } catch (Throwable t) {
        if (Configuration.getBoolean(PropertyKey.MASTER_JOURNAL_BACKUP_WHEN_CORRUPTED)) {
          takeEmergencyBackup();
        }
        throw t;
      }
      mLeaderSelector.waitForState(NodeState.STANDBY);
      mLastLosePrimacyTime = CommonUtils.getCurrentMs();
      if (Configuration.getBoolean(PropertyKey.MASTER_JOURNAL_EXIT_ON_DEMOTION)) {
        stop();
      } else {
        if (!mRunning) {
          break;
        }
        // Dump important information asynchronously
        ExecutorService es = null;
        List<Future<Void>> dumpFutures = new ArrayList<>();
        try {
          es = Executors.newFixedThreadPool(
              2, ThreadFactoryUtils.build("info-dumper-%d", true));
          dumpFutures.addAll(ProcessUtils.dumpInformationOnFailover(es));
        } catch (Throwable t) {
          LOG.warn("Failed to dump metrics and jstacks before demotion", t);
        }
        // Shut down services like RPC, WebServer, Journal and all master components
        LOG.info("Losing the leadership.");
        mServices.forEach(SimpleService::demote);
        demote();
        // Block until information dump is done and close resources
        for (Future<Void> f : dumpFutures) {
          try {
            f.get();
          } catch (InterruptedException | ExecutionException e) {
            LOG.warn("Failed to dump metrics and jstacks before demotion", e);
          }
        }
        if (es != null) {
          es.shutdownNow();
        }
      }
    }
  }
@Test
  public void startStopStandby() throws Exception {
    AlluxioMasterProcess master = new AlluxioMasterProcess(
        new NoopJournalSystem(), new AlwaysStandbyPrimarySelector());
    master.registerService(
        RpcServerService.Factory.create(master.getRpcBindAddress(), master, master.getRegistry()));
    master.registerService(WebServerService.Factory.create(master.getWebBindAddress(), master));
    master.registerService(MetricsService.Factory.create());
    Thread t = new Thread(() -> {
      try {
        master.start();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });
    t.start();
    startStopTest(master, false,
        Configuration.getBoolean(PropertyKey.STANDBY_MASTER_WEB_ENABLED),
        Configuration.getBoolean(PropertyKey.STANDBY_MASTER_METRICS_SINK_ENABLED));
  }