private Map<String, Path> copyHFilesToStagingDir() throws IOException {
    Map<String, Path> mapOfCopiedHFiles = new HashMap<>();
    Pair<byte[], List<String>> familyHFilePathsPair;
    List<String> hfilePaths;
    byte[] family;
    Path familyStagingDir;
    int familyHFilePathsPairsListSize;
    int totalNoOfHFiles;
    List<Pair<byte[], List<String>>> familyHFilePathsPairsList;
    FileSystem sourceFs = null;

    try {
      Path sourceClusterPath = new Path(sourceBaseNamespaceDirPath);
      /*
       * Path#getFileSystem will by default get the FS from cache. If both source and sink cluster
       * has same FS name service then it will return peer cluster FS. To avoid this we explicitly
       * disable the loading of FS from cache, so that a new FS is created with source cluster
       * configuration.
       */
      String sourceScheme = sourceClusterPath.toUri().getScheme();
      String disableCacheName =
        String.format("fs.%s.impl.disable.cache", new Object[] { sourceScheme });
      sourceClusterConf.setBoolean(disableCacheName, true);

      sourceFs = sourceClusterPath.getFileSystem(sourceClusterConf);

      User user = userProvider.getCurrent();
      // For each table name in the map
      for (Entry<String, List<Pair<byte[], List<String>>>> tableEntry : bulkLoadHFileMap
        .entrySet()) {
        String tableName = tableEntry.getKey();

        // Create staging directory for each table
        Path stagingDir = createStagingDir(hbaseStagingDir, user, TableName.valueOf(tableName));

        familyHFilePathsPairsList = tableEntry.getValue();
        familyHFilePathsPairsListSize = familyHFilePathsPairsList.size();

        // For each list of family hfile paths pair in the table
        for (int i = 0; i < familyHFilePathsPairsListSize; i++) {
          familyHFilePathsPair = familyHFilePathsPairsList.get(i);

          family = familyHFilePathsPair.getFirst();
          hfilePaths = familyHFilePathsPair.getSecond();

          familyStagingDir = new Path(stagingDir, Bytes.toString(family));
          totalNoOfHFiles = hfilePaths.size();

          // For each list of hfile paths for the family
          List<Future<Void>> futures = new ArrayList<>();
          Callable<Void> c;
          Future<Void> future;
          int currentCopied = 0;
          // Copy the hfiles parallely
          while (totalNoOfHFiles > currentCopied + this.copiesPerThread) {
            c = new Copier(sourceFs, familyStagingDir,
              hfilePaths.subList(currentCopied, currentCopied + this.copiesPerThread));
            future = exec.submit(c);
            futures.add(future);
            currentCopied += this.copiesPerThread;
          }

          int remaining = totalNoOfHFiles - currentCopied;
          if (remaining > 0) {
            c = new Copier(sourceFs, familyStagingDir,
              hfilePaths.subList(currentCopied, currentCopied + remaining));
            future = exec.submit(c);
            futures.add(future);
          }

          for (Future<Void> f : futures) {
            try {
              f.get();
            } catch (InterruptedException e) {
              InterruptedIOException iioe = new InterruptedIOException(
                "Failed to copy HFiles to local file system. This will be retried again "
                  + "by the source cluster.");
              iioe.initCause(e);
              throw iioe;
            } catch (ExecutionException e) {
              throw new IOException("Failed to copy HFiles to local file system. This will "
                + "be retried again by the source cluster.", e);
            }
          }
        }
        // Add the staging directory to this table. Staging directory contains all the hfiles
        // belonging to this table
        mapOfCopiedHFiles.put(tableName, stagingDir);
      }
      return mapOfCopiedHFiles;
    } finally {
      if (sourceFs != null) {
        sourceFs.close();
      }
      if (exec != null) {
        exec.shutdown();
      }
    }
  }
Number of hfiles to copy per thread during replication
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java