@Override
  public String backupTables(BackupRequest request) throws IOException {
    BackupType type = request.getBackupType();
    String targetRootDir = request.getTargetRootDir();
    List<TableName> tableList = request.getTableList();

    String backupId = BackupRestoreConstants.BACKUPID_PREFIX + EnvironmentEdgeManager.currentTime();
    if (type == BackupType.INCREMENTAL) {
      Set<TableName> incrTableSet;
      try (BackupSystemTable table = new BackupSystemTable(conn)) {
        incrTableSet = table.getIncrementalBackupTableSet(targetRootDir);
      }

      if (incrTableSet.isEmpty()) {
        String msg =
          "Incremental backup table set contains no tables. " + "You need to run full backup first "
            + (tableList != null ? "on " + StringUtils.join(tableList, ",") : "");

        throw new IOException(msg);
      }
      if (tableList != null) {
        tableList.removeAll(incrTableSet);
        if (!tableList.isEmpty()) {
          String extraTables = StringUtils.join(tableList, ",");
          String msg = "Some tables (" + extraTables + ") haven't gone through full backup. "
            + "Perform full backup on " + extraTables + " first, " + "then retry the command";
          throw new IOException(msg);
        }
      }
      tableList = Lists.newArrayList(incrTableSet);
    }
    if (tableList != null && !tableList.isEmpty()) {
      for (TableName table : tableList) {
        String targetTableBackupDir =
          HBackupFileSystem.getTableBackupDir(targetRootDir, backupId, table);
        Path targetTableBackupDirPath = new Path(targetTableBackupDir);
        FileSystem outputFs =
          FileSystem.get(targetTableBackupDirPath.toUri(), conn.getConfiguration());
        if (outputFs.exists(targetTableBackupDirPath)) {
          throw new IOException(
            "Target backup directory " + targetTableBackupDir + " exists already.");
        }
        outputFs.mkdirs(targetTableBackupDirPath);
      }
      ArrayList<TableName> nonExistingTableList = null;
      try (Admin admin = conn.getAdmin()) {
        for (TableName tableName : tableList) {
          if (!admin.tableExists(tableName)) {
            if (nonExistingTableList == null) {
              nonExistingTableList = new ArrayList<>();
            }
            nonExistingTableList.add(tableName);
          }
        }
      }
      if (nonExistingTableList != null) {
        if (type == BackupType.INCREMENTAL) {
          // Update incremental backup set
          tableList = excludeNonExistingTables(tableList, nonExistingTableList);
        } else {
          // Throw exception only in full mode - we try to backup non-existing table
          throw new IOException(
            "Non-existing tables found in the table list: " + nonExistingTableList);
        }
      }
    }

    // update table list
    BackupRequest.Builder builder = new BackupRequest.Builder();
    request = builder.withBackupType(request.getBackupType()).withTableList(tableList)
      .withTargetRootDir(request.getTargetRootDir()).withBackupSetName(request.getBackupSetName())
      .withTotalTasks(request.getTotalTasks()).withBandwidthPerTasks((int) request.getBandwidth())
      .withNoChecksumVerify(request.getNoChecksumVerify()).build();

    TableBackupClient client;
    try {
      client = BackupClientFactory.create(conn, backupId, request);
    } catch (IOException e) {
      LOG.error("There is an active session already running");
      throw e;
    }

    client.execute();

    return backupId;
  }
Clean up the data at target directory
@throws IOException if cleaning up the backup directory fails
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupAdminImpl.java