@Override
    public Job execute() throws Exception {

      // reflection preparation for private methods and fields
      Class<?> classDistCp = org.apache.hadoop.tools.DistCp.class;
      Method methodCleanup = classDistCp.getDeclaredMethod("cleanup");

      Field fieldInputOptions = getInputOptionsField(classDistCp);
      Field fieldSubmitted = classDistCp.getDeclaredField("submitted");

      methodCleanup.setAccessible(true);
      fieldInputOptions.setAccessible(true);
      fieldSubmitted.setAccessible(true);

      // execute() logic starts here
      assert fieldInputOptions.get(this) != null;

      Job job = null;
      try {

        List<Path> srcs = getSourcePaths(fieldInputOptions);

        long totalSrcLgth = 0;
        for (Path aSrc : srcs) {
          totalSrcLgth += BackupUtils.getFilesLength(aSrc.getFileSystem(super.getConf()), aSrc);
        }

        // Async call
        job = super.execute();
        // Update the copy progress to system table every 0.5s if progress value changed
        int progressReportFreq = MapReduceBackupCopyJob.this.getConf()
          .getInt("hbase.backup.progressreport.frequency", 500);
        float lastProgress = progressDone;
        while (!job.isComplete()) {
          float newProgress =
            progressDone + job.mapProgress() * subTaskPercntgInWholeTask * (1 - INIT_PROGRESS);

          if (newProgress > lastProgress) {

            BigDecimal progressData =
              new BigDecimal(newProgress * 100).setScale(1, BigDecimal.ROUND_HALF_UP);
            String newProgressStr = progressData + "%";
            LOG.info("Progress: " + newProgressStr);
            updateProgress(backupInfo, backupManager, progressData.intValue(), bytesCopied);
            LOG.debug("Backup progress data updated to backup system table: \"Progress: "
              + newProgressStr + ".\"");
            lastProgress = newProgress;
          }
          Thread.sleep(progressReportFreq);
        }
        // update the progress data after copy job complete
        float newProgress =
          progressDone + job.mapProgress() * subTaskPercntgInWholeTask * (1 - INIT_PROGRESS);
        BigDecimal progressData =
          new BigDecimal(newProgress * 100).setScale(1, BigDecimal.ROUND_HALF_UP);

        String newProgressStr = progressData + "%";
        LOG.info("Progress: " + newProgressStr + " subTask: " + subTaskPercntgInWholeTask
          + " mapProgress: " + job.mapProgress());

        // accumulate the overall backup progress
        progressDone = newProgress;
        bytesCopied += totalSrcLgth;

        updateProgress(backupInfo, backupManager, progressData.intValue(), bytesCopied);
        LOG.debug("Backup progress data updated to backup system table: \"Progress: "
          + newProgressStr + " - " + bytesCopied + " bytes copied.\"");
      } catch (Throwable t) {
        LOG.error(t.toString(), t);
        throw t;
      }

      String jobID = job.getJobID().toString();
      job.getConfiguration().set(DistCpConstants.CONF_LABEL_DISTCP_JOB_ID, jobID);

      LOG.debug(
        "DistCp job-id: " + jobID + " completed: " + job.isComplete() + " " + job.isSuccessful());
      Counters ctrs = job.getCounters();
      LOG.debug(Objects.toString(ctrs));
      if (job.isComplete() && !job.isSuccessful()) {
        throw new Exception("DistCp job-id: " + jobID + " failed");
      }

      return job;
    }
Extends DistCp for progress updating to backup system table during backup. Using DistCpV2
(MAPREDUCE-2765). Simply extend it and override execute() method to get the Job reference for
progress updating. Only the argument "src1, [src2, [...]] dst" is supported, no more DistCp
options.
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/mapreduce/MapReduceBackupCopyJob.java