code: @Override
    public void commitTask(TaskAttemptContext context) throws IOException {
      Path workPath = getWorkPath();
      FileSystem fs = workPath.getFileSystem(context.getConfiguration());
  
      if (fs.exists(workPath)) {
        long recordCount = getRecordCountFromCounter(context, RecordKeyDedupReducerBase.EVENT_COUNTER.RECORD_COUNT);
        String fileNamePrefix;
        if (recordCount == 0) {
  
          // recordCount == 0 indicates that it is a map-only, non-dedup job, and thus record count should
          // be obtained from mapper counter.
          fileNamePrefix = CompactionRecordCountProvider.M_OUTPUT_FILE_PREFIX;
          recordCount = getRecordCountFromCounter(context, RecordKeyMapperBase.EVENT_COUNTER.RECORD_COUNT);
        } else {
          fileNamePrefix = CompactionRecordCountProvider.MR_OUTPUT_FILE_PREFIX;
        }
        String fileName = CompactionRecordCountProvider.constructFileName(fileNamePrefix,
            "." + compactionFileExtension, recordCount);
  
        for (FileStatus status : fs.listStatus(workPath, new PathFilter() {
          @Override
          public boolean accept(Path path) {
            return FilenameUtils.isExtension(path.getName(), compactionFileExtension);
          }
        })) {
          Path newPath = new Path(status.getPath().getParent(), fileName);
          LOG.info(String.format("Renaming %s to %s", status.getPath(), newPath));
          fs.rename(status.getPath(), newPath);
        }
      }
  
      super.commitTask(context);
    }
docs: Commits the task, moving files to their final committed location by delegating to
  {@link FileOutputCommitter} to perform the actual moving. First, renames the
  files to include the count of records contained within the file and a timestamp,
  in the form {recordCount}.{timestamp}.avro. Then, the files are moved to their
  committed location.