public long splitLogDistributed(final Set<ServerName> serverNames, final List<Path> logDirs,
    PathFilter filter) throws IOException {
    MonitoredTask status = TaskMonitor.get()
      .createStatus("Doing distributed log split in " + logDirs + " for serverName=" + serverNames);
    long totalSize = 0;
    TaskBatch batch = null;
    long startTime = 0;
    List<FileStatus> logfiles = getFileList(logDirs, filter);
    if (!logfiles.isEmpty()) {
      status.setStatus("Checking directory contents...");
      SplitLogCounters.tot_mgr_log_split_batch_start.increment();
      LOG.info(
        "Started splitting " + logfiles.size() + " logs in " + logDirs + " for " + serverNames);
      startTime = EnvironmentEdgeManager.currentTime();
      batch = new TaskBatch();
      for (FileStatus lf : logfiles) {
        // TODO If the log file is still being written to - which is most likely
        // the case for the last log file - then its length will show up here
        // as zero. The size of such a file can only be retrieved after
        // recover-lease is done. totalSize will be under in most cases and the
        // metrics that it drives will also be under-reported.
        totalSize += lf.getLen();
        String pathToLog = CommonFSUtils.removeWALRootPath(lf.getPath(), conf);
        if (!enqueueSplitTask(pathToLog, batch)) {
          throw new IOException("duplicate log split scheduled for " + lf.getPath());
        }
      }
      waitForSplittingCompletion(batch, status);
    }

    if (batch != null && batch.done != batch.installed) {
      batch.isDead = true;
      SplitLogCounters.tot_mgr_log_split_batch_err.increment();
      LOG.warn("error while splitting logs in " + logDirs + " installed = " + batch.installed
        + " but only " + batch.done + " done");
      String msg = "error or interrupted while splitting logs in " + logDirs + " Task = " + batch;
      status.abort(msg);
      throw new IOException(msg);
    }
    for (Path logDir : logDirs) {
      status.setStatus("Cleaning up log directory...");
      final FileSystem fs = logDir.getFileSystem(conf);
      try {
        if (fs.exists(logDir) && !fs.delete(logDir, false)) {
          LOG.warn("Unable to delete log src dir. Ignoring. " + logDir);
        }
      } catch (IOException ioe) {
        FileStatus[] files = fs.listStatus(logDir);
        if (files != null && files.length > 0) {
          LOG.warn(
            "Returning success without actually splitting and "
              + "deleting all the log files in path " + logDir + ": " + Arrays.toString(files),
            ioe);
        } else {
          LOG.warn("Unable to delete log src dir. Ignoring. " + logDir, ioe);
        }
      }
      SplitLogCounters.tot_mgr_log_split_batch_success.increment();
    }
    String msg = "Finished splitting (more than or equal to) " + StringUtils.humanSize(totalSize)
      + " (" + totalSize + " bytes) in " + ((batch == null) ? 0 : batch.installed)
      + " log files in " + logDirs + " in "
      + ((startTime == 0) ? startTime : (EnvironmentEdgeManager.currentTime() - startTime)) + "ms";
    status.markComplete(msg);
    LOG.info(msg);
    return totalSize;
  }
The caller will block until all the hbase:meta log files of the given region server have been
processed - successfully split or an error is encountered - by an available worker region
server. This method must only be called after the region servers have been brought online.
@param logDirs List of log dirs to split
@param filter  the Path filter to select specific files for considering
@throws IOException If there was an error while splitting any log file
@return cumulative size of the logfiles split
hbase-server/src/main/java/org/apache/hadoop/hbase/master/SplitLogManager.java