@Override
  public void run(String[] backupIds) throws IOException {
    String bulkOutputConfKey;

    // TODO : run player on remote cluster
    player = new MapReduceHFileSplitterJob();
    bulkOutputConfKey = MapReduceHFileSplitterJob.BULK_OUTPUT_CONF_KEY;
    // Player reads all files in arbitrary directory structure and creates
    // a Map task for each file
    String bids = StringUtils.join(backupIds, ",");

    if (LOG.isDebugEnabled()) {
      LOG.debug("Merge backup images " + bids);
    }

    List<Pair<TableName, Path>> processedTableList = new ArrayList<>();
    boolean finishedTables = false;
    Connection conn = ConnectionFactory.createConnection(getConf());
    BackupSystemTable table = new BackupSystemTable(conn);
    FileSystem fs = null;

    try {

      // Get exclusive lock on backup system
      table.startBackupExclusiveOperation();
      // Start merge operation
      table.startMergeOperation(backupIds);

      // Select most recent backup id
      String mergedBackupId = BackupUtils.findMostRecentBackupId(backupIds);

      TableName[] tableNames = getTableNamesInBackupImages(backupIds);

      BackupInfo bInfo = table.readBackupInfo(backupIds[0]);
      String backupRoot = bInfo.getBackupRootDir();
      Path backupRootPath = new Path(backupRoot);
      fs = backupRootPath.getFileSystem(conf);

      for (int i = 0; i < tableNames.length; i++) {
        LOG.info("Merge backup images for " + tableNames[i]);

        // Find input directories for table
        Path[] dirPaths = findInputDirectories(fs, backupRoot, tableNames[i], backupIds);
        String dirs = StringUtils.join(dirPaths, ",");

        // bulkOutputPath should be on the same filesystem as backupRoot
        Path tmpRestoreOutputDir = HBackupFileSystem.getBackupTmpDirPath(backupRoot);
        Path bulkOutputPath = BackupUtils.getBulkOutputDir(tmpRestoreOutputDir,
          BackupUtils.getFileNameCompatibleString(tableNames[i]), getConf(), false);
        // Delete content if exists
        if (fs.exists(bulkOutputPath)) {
          if (!fs.delete(bulkOutputPath, true)) {
            LOG.warn("Can not delete: " + bulkOutputPath);
          }
        }
        Configuration conf = getConf();
        conf.set(bulkOutputConfKey, bulkOutputPath.toString());
        String[] playerArgs = { dirs, tableNames[i].getNameAsString() };

        player.setConf(getConf());
        int result = player.run(playerArgs);
        if (!succeeded(result)) {
          throw new IOException("Can not merge backup images for " + dirs
            + " (check Hadoop/MR and HBase logs). Player return code =" + result);
        }
        // Add to processed table list
        processedTableList.add(new Pair<>(tableNames[i], bulkOutputPath));
        LOG.debug("Merge Job finished:" + result);
      }
      List<TableName> tableList = toTableNameList(processedTableList);
      table.updateProcessedTablesForMerge(tableList);
      finishedTables = true;

      // PHASE 2 (modification of a backup file system)
      // Move existing mergedBackupId data into tmp directory
      // we will need it later in case of a failure
      Path tmpBackupDir =
        HBackupFileSystem.getBackupTmpDirPathForBackupId(backupRoot, mergedBackupId);
      Path backupDirPath = HBackupFileSystem.getBackupPath(backupRoot, mergedBackupId);

      if (!fs.rename(backupDirPath, tmpBackupDir)) {
        throw new IOException("Failed to rename " + backupDirPath + " to " + tmpBackupDir);
      } else {
        LOG.debug("Renamed " + backupDirPath + " to " + tmpBackupDir);
      }
      // Move new data into backup dest
      for (Pair<TableName, Path> tn : processedTableList) {
        moveData(fs, backupRoot, tn.getSecond(), tn.getFirst(), mergedBackupId);
      }
      // Update backup manifest
      List<String> backupsToDelete = getBackupIdsToDelete(backupIds, mergedBackupId);
      updateBackupManifest(tmpBackupDir.getParent().toString(), mergedBackupId, backupsToDelete);
      // Copy meta files back from tmp to backup dir
      copyMetaData(fs, tmpBackupDir, backupDirPath);
      // Delete tmp dir (Rename back during repair)
      if (!fs.delete(tmpBackupDir, true)) {
        // WARN and ignore
        LOG.warn("Could not delete tmp dir: " + tmpBackupDir);
      }
      // Delete old data
      deleteBackupImages(backupsToDelete, conn, fs, backupRoot);
      // Finish merge session
      table.finishMergeOperation();
      // Release lock
      table.finishBackupExclusiveOperation();
    } catch (RuntimeException e) {

      throw e;
    } catch (Exception e) {
      LOG.error(e.toString(), e);
      if (!finishedTables) {
        // cleanup bulk directories and finish merge
        // merge MUST be repeated (no need for repair)
        if (fs != null) {
          cleanupBulkLoadDirs(fs, toPathList(processedTableList));
        }
        table.finishMergeOperation();
        table.finishBackupExclusiveOperation();
        throw new IOException("Backup merge operation failed, you should try it again", e);
      } else {
        // backup repair must be run
        throw new IOException(
          "Backup merge operation failed, run backup repair tool to restore system's integrity", e);
      }
    } finally {
      table.close();
      conn.close();
    }
  }
MapReduce implementation of {@link BackupMergeJob} Must be initialized with configuration of a
backup destination cluster
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/mapreduce/MapReduceBackupMergeJob.java