@Override
  protected Flow executeFromState(final MasterProcedureEnv env, final CloneSnapshotState state)
    throws InterruptedException {
    LOG.trace("{} execute state={}", this, state);
    try {
      switch (state) {
        case CLONE_SNAPSHOT_PRE_OPERATION:
          // Verify if we can clone the table
          prepareClone(env);

          preCloneSnapshot(env);
          setNextState(CloneSnapshotState.CLONE_SNAPSHOT_WRITE_FS_LAYOUT);
          break;
        case CLONE_SNAPSHOT_WRITE_FS_LAYOUT:
          updateTableDescriptorWithSFT();
          newRegions = createFilesystemLayout(env, tableDescriptor, newRegions);
          env.getMasterServices().getTableDescriptors().update(tableDescriptor, true);
          if (tableDescriptor.getErasureCodingPolicy() != null) {
            setNextState(CloneSnapshotState.CLONE_SNAPSHOT_SET_ERASURE_CODING_POLICY);
          } else {
            setNextState(CloneSnapshotState.CLONE_SNAPSHOT_ADD_TO_META);
          }
          break;
        case CLONE_SNAPSHOT_SET_ERASURE_CODING_POLICY:
          ErasureCodingUtils.setPolicy(env.getMasterFileSystem().getFileSystem(),
            env.getMasterFileSystem().getRootDir(), getTableName(),
            tableDescriptor.getErasureCodingPolicy());
          setNextState(CloneSnapshotState.CLONE_SNAPSHOT_ADD_TO_META);
          break;
        case CLONE_SNAPSHOT_ADD_TO_META:
          addRegionsToMeta(env);
          setNextState(CloneSnapshotState.CLONE_SNAPSHOT_ASSIGN_REGIONS);
          break;
        case CLONE_SNAPSHOT_ASSIGN_REGIONS:
          CreateTableProcedure.setEnablingState(env, getTableName());

          // Separate newRegions to split regions and regions to assign
          List<RegionInfo> splitRegions = new ArrayList<>();
          List<RegionInfo> regionsToAssign = new ArrayList<>();
          newRegions.forEach(ri -> {
            if (ri.isOffline() && (ri.isSplit() || ri.isSplitParent())) {
              splitRegions.add(ri);
            } else {
              regionsToAssign.add(ri);
            }
          });

          // For split regions, add them to RegionStates
          AssignmentManager am = env.getAssignmentManager();
          splitRegions
            .forEach(ri -> am.getRegionStates().updateRegionState(ri, RegionState.State.SPLIT));

          addChildProcedure(
            env.getAssignmentManager().createRoundRobinAssignProcedures(regionsToAssign));
          setNextState(CloneSnapshotState.CLONE_SNAPSHOT_UPDATE_DESC_CACHE);
          break;
        case CLONE_SNAPSHOT_UPDATE_DESC_CACHE:
          // XXX: this stage should be named as set table enabled, as now we will cache the
          // descriptor after writing fs layout.
          CreateTableProcedure.setEnabledState(env, getTableName());
          setNextState(CloneSnapshotState.CLONE_SNAPHOST_RESTORE_ACL);
          break;
        case CLONE_SNAPHOST_RESTORE_ACL:
          restoreSnapshotAcl(env);
          setNextState(CloneSnapshotState.CLONE_SNAPSHOT_POST_OPERATION);
          break;
        case CLONE_SNAPSHOT_POST_OPERATION:
          postCloneSnapshot(env);

          MetricsSnapshot metricsSnapshot = new MetricsSnapshot();
          metricsSnapshot.addSnapshotClone(
            getMonitorStatus().getCompletionTimestamp() - getMonitorStatus().getStartTime());
          getMonitorStatus().markComplete("Clone snapshot '" + snapshot.getName() + "' completed!");
          return Flow.NO_MORE_STATE;
        default:
          throw new UnsupportedOperationException("unhandled state=" + state);
      }
    } catch (IOException e) {
      if (isRollbackSupported(state)) {
        setFailure("master-clone-snapshot", e);
      } else {
        LOG.warn("Retriable error trying to clone snapshot=" + snapshot.getName() + " to table="
          + getTableName() + " state=" + state, e);
      }
    }
    return Flow.HAS_MORE_STATE;
  }
Set up monitor status if it is not created.
hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/CloneSnapshotProcedure.java