@Override
  protected Flow executeFromState(MasterProcedureEnv env, PeerModificationState state)
    throws ProcedureSuspendedException, InterruptedException {
    switch (state) {
      case PRE_PEER_MODIFICATION:
        try {
          if (shouldFailForMigrating(env)) {
            LOG.info("There is a pending {}, give up execution of {}",
              MigrateReplicationQueueFromZkToTableProcedure.class.getSimpleName(),
              getClass().getName());
            setFailure("master-" + getPeerOperationType().name().toLowerCase() + "-peer",
              new DoNotRetryIOException("There is a pending "
                + MigrateReplicationQueueFromZkToTableProcedure.class.getSimpleName()));
            releaseLatch(env);
            return Flow.NO_MORE_STATE;
          }
          checkPeerModificationEnabled(env);
          prePeerModification(env);
        } catch (IOException e) {
          LOG.warn("{} failed to call pre CP hook or the pre check is failed for peer {}, "
            + "mark the procedure as failure and give up", getClass().getName(), peerId, e);
          setFailure("master-" + getPeerOperationType().name().toLowerCase() + "-peer", e);
          releaseLatch(env);
          return Flow.NO_MORE_STATE;
        } catch (ReplicationException e) {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn("{} failed to call prePeerModification for peer {}, sleep {} secs",
              getClass().getName(), peerId, backoff / 1000, e));
        }
        resetRetry();
        setNextState(PeerModificationState.UPDATE_PEER_STORAGE);
        return Flow.HAS_MORE_STATE;
      case UPDATE_PEER_STORAGE:
        try {
          updatePeerStorage(env);
        } catch (ReplicationException e) {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn("{} update peer storage for peer {} failed, sleep {} secs",
              getClass().getName(), peerId, backoff / 1000, e));
        }
        resetRetry();
        setNextState(PeerModificationState.REFRESH_PEER_ON_RS);
        return Flow.HAS_MORE_STATE;
      case REFRESH_PEER_ON_RS:
        refreshPeer(env, getPeerOperationType());
        setNextState(nextStateAfterRefresh());
        return Flow.HAS_MORE_STATE;
      case SERIAL_PEER_REOPEN_REGIONS:
        try {
          reopenRegions(env);
        } catch (Exception e) {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn("{} reopen regions for peer {} failed,  sleep {} secs",
              getClass().getName(), peerId, backoff / 1000, e));
        }
        resetRetry();
        setNextState(PeerModificationState.SERIAL_PEER_UPDATE_LAST_PUSHED_SEQ_ID);
        return Flow.HAS_MORE_STATE;
      case SERIAL_PEER_UPDATE_LAST_PUSHED_SEQ_ID:
        try {
          updateLastPushedSequenceIdForSerialPeer(env);
        } catch (Exception e) {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn("{} set last sequence id for peer {} failed,  sleep {} secs",
              getClass().getName(), peerId, backoff / 1000, e));
        }
        resetRetry();
        setNextState(enablePeerBeforeFinish()
          ? PeerModificationState.SERIAL_PEER_SET_PEER_ENABLED
          : PeerModificationState.POST_PEER_MODIFICATION);
        return Flow.HAS_MORE_STATE;
      case SERIAL_PEER_SET_PEER_ENABLED:
        try {
          enablePeer(env);
        } catch (ReplicationException e) {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn("{} enable peer before finish for peer {} failed,  sleep {} secs",
              getClass().getName(), peerId, backoff / 1000, e));
        }
        resetRetry();
        setNextState(PeerModificationState.SERIAL_PEER_ENABLE_PEER_REFRESH_PEER_ON_RS);
        return Flow.HAS_MORE_STATE;
      case SERIAL_PEER_ENABLE_PEER_REFRESH_PEER_ON_RS:
        refreshPeer(env, PeerOperationType.ENABLE);
        setNextState(PeerModificationState.POST_PEER_MODIFICATION);
        return Flow.HAS_MORE_STATE;
      case POST_PEER_MODIFICATION:
        try {
          postPeerModification(env);
        } catch (ReplicationException e) {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn(
              "{} failed to call postPeerModification for peer {},  sleep {} secs",
              getClass().getName(), peerId, backoff / 1000, e));
        } catch (IOException e) {
          LOG.warn("{} failed to call post CP hook for peer {}, "
            + "ignore since the procedure has already done", getClass().getName(), peerId, e);
        }
        releaseLatch(env);
        return Flow.NO_MORE_STATE;
      default:
        throw new UnsupportedOperationException("unhandled state=" + state);
    }
  }
The implementation class should override this method if the procedure may enter the serial
related states.
hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ModifyPeerProcedure.java