class HDFSSynchronization implements ExternalSynchronization {

  @Override
  public TaskAttemptID acquireTaskAttemptIdLock(Configuration conf, int taskId) {
    String jobJtIdentifier = getJobJtIdentifier(conf);
    JobID jobId = HadoopFormats.getJobId(conf);
    int taskAttemptCandidate = 0;
    boolean taskAttemptAcquired = false;

    while (!taskAttemptAcquired) {
      taskAttemptCandidate++;
      Path path =
          new Path(
              locksDir,
              String.format(
                  LOCKS_DIR_TASK_ATTEMPT_PATTERN, jobJtIdentifier, taskId, taskAttemptCandidate));
      taskAttemptAcquired = tryCreateFile(conf, path);
    }

    return HadoopFormats.createTaskAttemptID(jobId, taskId, taskAttemptCandidate);
  }

  public  HDFSSynchronization(String locksDir);
    HDFSSynchronization(
      String locksDir, ThrowingFunction<Configuration, FileSystem, IOException> fileSystemFactory);

}