public Job createSubmittableJob(Configuration conf, String[] args) throws IOException {
    if (!doCommandLine(args)) {
      return null;
    }
    conf.set(NAME + ".tableName", tableName);
    conf.setLong(NAME + ".startTime", startTime);
    conf.setLong(NAME + ".endTime", endTime);
    conf.setInt(NAME + ".sleepMsBeforeReCompare", sleepMsBeforeReCompare);
    conf.set(NAME + ".delimiter", delimiter);
    conf.setInt(NAME + ".batch", batch);
    conf.setBoolean(NAME + ".verbose", verbose);
    conf.setBoolean(NAME + ".includeDeletedCells", includeDeletedCells);
    if (families != null) {
      conf.set(NAME + ".families", families);
    }
    if (rowPrefixes != null) {
      conf.set(NAME + ".rowPrefixes", rowPrefixes);
    }

    String peerQuorumAddress;
    Pair<ReplicationPeerConfig, Configuration> peerConfigPair = null;
    if (peerId != null) {
      peerConfigPair = getPeerQuorumConfig(conf, peerId);
      ReplicationPeerConfig peerConfig = peerConfigPair.getFirst();
      peerQuorumAddress = peerConfig.getClusterKey();
      LOG.info("Peer Quorum Address: " + peerQuorumAddress + ", Peer Configuration: "
        + peerConfig.getConfiguration());
      conf.set(NAME + ".peerQuorumAddress", peerQuorumAddress);
      HBaseConfiguration.setWithPrefix(conf, PEER_CONFIG_PREFIX,
        peerConfig.getConfiguration().entrySet());
    } else {
      assert this.peerQuorumAddress != null;
      peerQuorumAddress = this.peerQuorumAddress;
      LOG.info("Peer Quorum Address: " + peerQuorumAddress);
      conf.set(NAME + ".peerQuorumAddress", peerQuorumAddress);
    }

    if (peerTableName != null) {
      LOG.info("Peer Table Name: " + peerTableName);
      conf.set(NAME + ".peerTableName", peerTableName);
    }

    conf.setInt(NAME + ".versions", versions);
    LOG.info("Number of version: " + versions);

    conf.setInt(NAME + ".recompareTries", reCompareTries);
    conf.setInt(NAME + ".recompareBackoffExponent", reCompareBackoffExponent);
    conf.setInt(NAME + ".recompareThreads", reCompareThreads);

    // Set Snapshot specific parameters
    if (peerSnapshotName != null) {
      conf.set(NAME + ".peerSnapshotName", peerSnapshotName);

      // for verifyRep by snapshot, choose a unique sub-directory under peerSnapshotTmpDir to
      // restore snapshot.
      Path restoreDir = new Path(peerSnapshotTmpDir, UUID.randomUUID().toString());
      peerSnapshotTmpDir = restoreDir.toString();
      conf.set(NAME + ".peerSnapshotTmpDir", peerSnapshotTmpDir);

      conf.set(NAME + ".peerFSAddress", peerFSAddress);
      conf.set(NAME + ".peerHBaseRootAddress", peerHBaseRootAddress);

      // This is to create HDFS delegation token for peer cluster in case of secured
      conf.setStrings(MRJobConfig.JOB_NAMENODES, peerFSAddress, conf.get(HConstants.HBASE_DIR));
    }

    Job job = Job.getInstance(conf, conf.get(JOB_NAME_CONF_KEY, NAME + "_" + tableName));
    job.setJarByClass(VerifyReplication.class);

    Scan scan = new Scan();
    scan.setTimeRange(startTime, endTime);
    scan.setRaw(includeDeletedCells);
    scan.setCacheBlocks(false);
    if (batch > 0) {
      scan.setBatch(batch);
    }
    if (versions >= 0) {
      scan.readVersions(versions);
      LOG.info("Number of versions set to " + versions);
    }
    if (families != null) {
      String[] fams = families.split(",");
      for (String fam : fams) {
        scan.addFamily(Bytes.toBytes(fam));
      }
    }

    setRowPrefixFilter(scan, rowPrefixes);

    if (sourceSnapshotName != null) {
      Path snapshotTempPath = new Path(sourceSnapshotTmpDir);
      LOG.info(
        "Using source snapshot-" + sourceSnapshotName + " with temp dir:" + sourceSnapshotTmpDir);
      TableMapReduceUtil.initTableSnapshotMapperJob(sourceSnapshotName, scan, Verifier.class, null,
        null, job, true, snapshotTempPath);
      restoreSnapshotForPeerCluster(conf, peerQuorumAddress);
    } else {
      TableMapReduceUtil.initTableMapperJob(tableName, scan, Verifier.class, null, null, job);
    }

    Configuration peerClusterBaseConf;
    if (peerId != null) {
      assert peerConfigPair != null;
      peerClusterBaseConf = peerConfigPair.getSecond();
    } else {
      peerClusterBaseConf = conf;
    }
    Configuration peerClusterConf;
    URI uri = ConnectionRegistryFactory.tryParseAsConnectionURI(peerQuorumAddress);
    if (uri != null) {
      peerClusterConf = new Configuration(peerClusterBaseConf);
      applyURIConf(peerClusterConf, uri);
    } else {
      peerClusterConf = HBaseConfiguration.createClusterConf(peerClusterBaseConf, peerQuorumAddress,
        PEER_CONFIG_PREFIX);
    }
    // Obtain the auth token from peer cluster
    TableMapReduceUtil.initCredentialsForCluster(job, peerClusterConf, uri);

    job.setOutputFormatClass(NullOutputFormat.class);
    job.setNumReduceTasks(0);
    return job;
  }
Sets up the actual job.
@param conf The current configuration.
@param args The command line parameters.
@return The newly created job.
@throws java.io.IOException When setting up the job fails.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/replication/VerifyReplication.java