@Override
  protected Flow executeFromState(MasterProcedureEnv env,
    MigrateReplicationQueueFromZkToTableState state)
    throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
    switch (state) {
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_DISABLE_CLEANER:
        disableReplicationLogCleaner(env);
        setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_PREPARE);
        return Flow.HAS_MORE_STATE;
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_PREPARE:
        waitUntilNoPeerProcedure(env);
        List<ReplicationPeerDescription> peers = env.getReplicationPeerManager().listPeers(null);
        if (peers.isEmpty()) {
          // we will not load the region_replica_replication peer, so here we need to check the
          // storage directly
          try {
            if (env.getReplicationPeerManager().hasRegionReplicaReplicationPeer()) {
              LOG.info(
                "No active replication peer found but we still have '{}' peer, need to"
                  + "wait until all region servers are upgraded",
                ServerRegionReplicaUtil.REGION_REPLICA_REPLICATION_PEER);
              setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_WAIT_UPGRADING);
              return Flow.HAS_MORE_STATE;
            }
          } catch (ReplicationException e) {
            throw suspend(env.getMasterConfiguration(), backoff -> LOG
              .warn("failed to list peer ids, sleep {} secs and retry later", backoff / 1000, e));
          }
          LOG.info("No active replication peer found, just clean up all replication queue data");
          setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_ENABLE_CLEANER);
          return Flow.HAS_MORE_STATE;
        }
        disabledPeerIds = peers.stream().filter(ReplicationPeerDescription::isEnabled)
          .map(ReplicationPeerDescription::getPeerId).collect(Collectors.toList());
        setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_DISABLE_PEER);
        resetRetry();
        return Flow.HAS_MORE_STATE;
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_DISABLE_PEER:
        for (String peerId : disabledPeerIds) {
          addChildProcedure(new DisablePeerProcedure(peerId));
        }
        setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_MIGRATE);
        return Flow.HAS_MORE_STATE;
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_MIGRATE:
        try {
          if (
            ProcedureFutureUtil.checkFuture(this, this::getFuture, this::setFuture,
              this::finishMigartion)
          ) {
            return Flow.HAS_MORE_STATE;
          }
          ProcedureFutureUtil.suspendIfNecessary(this, this::setFuture,
            env.getReplicationPeerManager()
              .migrateQueuesFromZk(env.getMasterServices().getZooKeeper(), getExecutorService()),
            env, this::finishMigartion);
        } catch (IOException e) {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn("failed to migrate queue data, sleep {} secs and retry later",
              backoff / 1000, e));
        }
        return Flow.HAS_MORE_STATE;
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_WAIT_UPGRADING:
        long rsWithLowerVersion =
          env.getMasterServices().getServerManager().getOnlineServers().values().stream()
            .filter(sm -> VersionInfo.getMajorVersion(sm.getVersion()) < MIN_MAJOR_VERSION).count();
        if (rsWithLowerVersion == 0) {
          setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_ENABLE_PEER);
          return Flow.HAS_MORE_STATE;
        } else {
          throw suspend(env.getMasterConfiguration(),
            backoff -> LOG.warn(
              "There are still {} region servers which have a major version"
                + " less than {}, sleep {} secs and check later",
              rsWithLowerVersion, MIN_MAJOR_VERSION, backoff / 1000));
        }
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_ENABLE_PEER:
        if (CollectionUtils.isNotEmpty(disabledPeerIds)) {
          for (String peerId : disabledPeerIds) {
            addChildProcedure(new EnablePeerProcedure(peerId));
          }
        }
        setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_ENABLE_CLEANER);
        return Flow.HAS_MORE_STATE;
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_ENABLE_CLEANER:
        enableReplicationLogCleaner(env);
        setNextState(MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_CLEAN_UP);
        return Flow.HAS_MORE_STATE;
      case MIGRATE_REPLICATION_QUEUE_FROM_ZK_TO_TABLE_CLEAN_UP:
        // this is mainly for deleting the region replica replication queue data, but anyway, since
        // we should have migrated all data, here we can simply delete everything
        cleanup(env);
        return Flow.NO_MORE_STATE;
      default:
        throw new UnsupportedOperationException("unhandled state=" + state);
    }
  }
A procedure for migrating replication queue data from zookeeper to hbase:replication table.
hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/MigrateReplicationQueueFromZkToTableProcedure.java