@Override
  public int doWork() throws IOException {
    Configuration conf = getConf();

    // Check user options
    if (snapshotName == null) {
      System.err.println("Snapshot name not provided.");
      LOG.error("Use -h or --help for usage instructions.");
      return EXIT_FAILURE;
    }

    if (outputRoot == null) {
      System.err
        .println("Destination file-system (--" + Options.COPY_TO.getLongOpt() + ") not provided.");
      LOG.error("Use -h or --help for usage instructions.");
      return EXIT_FAILURE;
    }

    if (targetName == null) {
      targetName = snapshotName;
    }
    if (inputRoot == null) {
      inputRoot = CommonFSUtils.getRootDir(conf);
    } else {
      CommonFSUtils.setRootDir(conf, inputRoot);
    }

    Configuration srcConf = HBaseConfiguration.createClusterConf(conf, null, CONF_SOURCE_PREFIX);
    FileSystem inputFs = FileSystem.get(inputRoot.toUri(), srcConf);
    Configuration destConf = HBaseConfiguration.createClusterConf(conf, null, CONF_DEST_PREFIX);
    FileSystem outputFs = FileSystem.get(outputRoot.toUri(), destConf);
    boolean skipTmp = conf.getBoolean(CONF_SKIP_TMP, false)
      || conf.get(SnapshotDescriptionUtils.SNAPSHOT_WORKING_DIR) != null;
    Path snapshotDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(snapshotName, inputRoot);
    Path snapshotTmpDir =
      SnapshotDescriptionUtils.getWorkingSnapshotDir(targetName, outputRoot, destConf);
    Path outputSnapshotDir =
      SnapshotDescriptionUtils.getCompletedSnapshotDir(targetName, outputRoot);
    Path initialOutputSnapshotDir = skipTmp ? outputSnapshotDir : snapshotTmpDir;
    LOG.debug("inputFs={}, inputRoot={}", inputFs.getUri().toString(), inputRoot);
    LOG.debug("outputFs={}, outputRoot={}, skipTmp={}, initialOutputSnapshotDir={}", outputFs,
      outputRoot.toString(), skipTmp, initialOutputSnapshotDir);

    // throw CorruptedSnapshotException if we can't read the snapshot info.
    SnapshotDescription sourceSnapshotDesc =
      SnapshotDescriptionUtils.readSnapshotInfo(inputFs, snapshotDir);

    // Verify snapshot source before copying files
    if (verifySource) {
      LOG.info("Verify the source snapshot's expiration status and integrity.");
      verifySnapshot(sourceSnapshotDesc, srcConf, inputFs, inputRoot, snapshotDir);
    }

    // Find the necessary directory which need to change owner and group
    Path needSetOwnerDir = SnapshotDescriptionUtils.getSnapshotRootDir(outputRoot);
    if (outputFs.exists(needSetOwnerDir)) {
      if (skipTmp) {
        needSetOwnerDir = outputSnapshotDir;
      } else {
        needSetOwnerDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(outputRoot, destConf);
        if (outputFs.exists(needSetOwnerDir)) {
          needSetOwnerDir = snapshotTmpDir;
        }
      }
    }

    // Check if the snapshot already exists
    if (outputFs.exists(outputSnapshotDir)) {
      if (overwrite) {
        if (!outputFs.delete(outputSnapshotDir, true)) {
          System.err.println("Unable to remove existing snapshot directory: " + outputSnapshotDir);
          return EXIT_FAILURE;
        }
      } else {
        System.err.println("The snapshot '" + targetName + "' already exists in the destination: "
          + outputSnapshotDir);
        return EXIT_FAILURE;
      }
    }

    if (!skipTmp) {
      // Check if the snapshot already in-progress
      if (outputFs.exists(snapshotTmpDir)) {
        if (overwrite) {
          if (!outputFs.delete(snapshotTmpDir, true)) {
            System.err
              .println("Unable to remove existing snapshot tmp directory: " + snapshotTmpDir);
            return EXIT_FAILURE;
          }
        } else {
          System.err
            .println("A snapshot with the same name '" + targetName + "' may be in-progress");
          System.err
            .println("Please check " + snapshotTmpDir + ". If the snapshot has completed, ");
          System.err
            .println("consider removing " + snapshotTmpDir + " by using the -overwrite option");
          return EXIT_FAILURE;
        }
      }
    }

    // Step 1 - Copy fs1:/.snapshot/<snapshot> to fs2:/.snapshot/.tmp/<snapshot>
    // The snapshot references must be copied before the hfiles otherwise the cleaner
    // will remove them because they are unreferenced.
    List<Path> travesedPaths = new ArrayList<>();
    boolean copySucceeded = false;
    try {
      LOG.info("Copy Snapshot Manifest from " + snapshotDir + " to " + initialOutputSnapshotDir);
      travesedPaths =
        FSUtils.copyFilesParallel(inputFs, snapshotDir, outputFs, initialOutputSnapshotDir, conf,
          conf.getInt(CONF_COPY_MANIFEST_THREADS, DEFAULT_COPY_MANIFEST_THREADS));
      copySucceeded = true;
    } catch (IOException e) {
      throw new ExportSnapshotException("Failed to copy the snapshot directory: from=" + snapshotDir
        + " to=" + initialOutputSnapshotDir, e);
    } finally {
      if (copySucceeded) {
        if (filesUser != null || filesGroup != null) {
          LOG.warn(
            (filesUser == null ? "" : "Change the owner of " + needSetOwnerDir + " to " + filesUser)
              + (filesGroup == null
                ? ""
                : ", Change the group of " + needSetOwnerDir + " to " + filesGroup));
          setOwnerParallel(outputFs, filesUser, filesGroup, conf, travesedPaths);
        }
        if (filesMode > 0) {
          LOG.warn("Change the permission of " + needSetOwnerDir + " to " + filesMode);
          setPermissionParallel(outputFs, (short) filesMode, travesedPaths, conf);
        }
      }
    }

    // Write a new .snapshotinfo if the target name is different from the source name or we want to
    // reset TTL for target snapshot.
    if (!targetName.equals(snapshotName) || resetTtl) {
      SnapshotDescription.Builder snapshotDescBuilder =
        SnapshotDescriptionUtils.readSnapshotInfo(inputFs, snapshotDir).toBuilder();
      if (!targetName.equals(snapshotName)) {
        snapshotDescBuilder.setName(targetName);
      }
      if (resetTtl) {
        snapshotDescBuilder.setTtl(HConstants.DEFAULT_SNAPSHOT_TTL);
      }
      SnapshotDescriptionUtils.writeSnapshotInfo(snapshotDescBuilder.build(),
        initialOutputSnapshotDir, outputFs);
      if (filesUser != null || filesGroup != null) {
        outputFs.setOwner(
          new Path(initialOutputSnapshotDir, SnapshotDescriptionUtils.SNAPSHOTINFO_FILE), filesUser,
          filesGroup);
      }
      if (filesMode > 0) {
        outputFs.setPermission(
          new Path(initialOutputSnapshotDir, SnapshotDescriptionUtils.SNAPSHOTINFO_FILE),
          new FsPermission((short) filesMode));
      }
    }

    // Step 2 - Start MR Job to copy files
    // The snapshot references must be copied before the files otherwise the files gets removed
    // by the HFileArchiver, since they have no references.
    try {
      runCopyJob(inputRoot, outputRoot, snapshotName, snapshotDir, verifyChecksum, filesUser,
        filesGroup, filesMode, mappers, bandwidthMB, storagePolicy);

      LOG.info("Finalize the Snapshot Export");
      if (!skipTmp) {
        // Step 3 - Rename fs2:/.snapshot/.tmp/<snapshot> fs2:/.snapshot/<snapshot>
        if (!outputFs.rename(snapshotTmpDir, outputSnapshotDir)) {
          throw new ExportSnapshotException("Unable to rename snapshot directory from="
            + snapshotTmpDir + " to=" + outputSnapshotDir);
        }
      }

      // Step 4 - Verify snapshot integrity
      if (verifyTarget) {
        LOG.info("Verify the exported snapshot's expiration status and integrity.");
        SnapshotDescription targetSnapshotDesc =
          SnapshotDescriptionUtils.readSnapshotInfo(outputFs, outputSnapshotDir);
        verifySnapshot(targetSnapshotDesc, destConf, outputFs, outputRoot, outputSnapshotDir);
      }

      LOG.info("Export Completed: " + targetName);
      return EXIT_SUCCESS;
    } catch (Exception e) {
      LOG.error("Snapshot export failed", e);
      if (!skipTmp) {
        outputFs.delete(snapshotTmpDir, true);
      }
      outputFs.delete(outputSnapshotDir, true);
      return EXIT_FAILURE;
    }
  }
Execute the export snapshot by copying the snapshot metadata, hfiles and wals.
@return 0 on success, and != 0 upon failure.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/snapshot/ExportSnapshot.java