@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);
  }
@Test
  public void testTaskAttemptIdAcquire() {
    int tasksCount = 100;
    int taskId = 25;

    for (int i = 0; i < tasksCount; i++) {
      TaskAttemptID taskAttemptID = tested.acquireTaskAttemptIdLock(configuration, taskId);
      assertTrue(isFileExists(getTaskAttemptIdPath(taskId, taskAttemptID.getId())));
    }
  }