code: public static void cleanTaskStagingData(State state, Logger logger, Closer closer,
        Map<String, ParallelRunner> parallelRunners) throws IOException {
      int numBranches = state.getPropAsInt(ConfigurationKeys.FORK_BRANCHES_KEY, 1);
  
      int parallelRunnerThreads =
          state.getPropAsInt(ParallelRunner.PARALLEL_RUNNER_THREADS_KEY, ParallelRunner.DEFAULT_PARALLEL_RUNNER_THREADS);
  
      for (int branchId = 0; branchId < numBranches; branchId++) {
        String writerFsUri = state.getProp(
            ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_SYSTEM_URI, numBranches, branchId),
            ConfigurationKeys.LOCAL_FS_URI);
        FileSystem fs = getFsWithProxy(state, writerFsUri, WriterUtils.getFsConfiguration(state));
        ParallelRunner parallelRunner = getParallelRunner(fs, closer, parallelRunnerThreads, parallelRunners);
  
        Path stagingPath = WriterUtils.getWriterStagingDir(state, numBranches, branchId);
        if (fs.exists(stagingPath)) {
          logger.info("Cleaning up staging directory " + stagingPath.toUri().getPath());
          parallelRunner.deletePath(stagingPath, true);
        }
  
        Path outputPath = WriterUtils.getWriterOutputDir(state, numBranches, branchId);
        if (fs.exists(outputPath)) {
          logger.info("Cleaning up output directory " + outputPath.toUri().getPath());
          parallelRunner.deletePath(outputPath, true);
        }
      }
    }
docs: Cleanup staging data of a Gobblin task using a {@link ParallelRunner}.
  
  @param state workunit state.
  @param logger a {@link Logger} used for logging.
  @param closer a closer that registers the given map of ParallelRunners. The caller is responsible
  for closing the closer after the cleaning is done.
  @param parallelRunners a map from FileSystem URI to ParallelRunner.
  @throws IOException if it fails to cleanup the task staging data.