@Override
  public TaskID acquireTaskIdLock(Configuration conf) {
    JobID jobId = HadoopFormats.getJobId(conf);
    boolean lockAcquired = false;
    int taskIdCandidate = 0;

    while (!lockAcquired) {
      taskIdCandidate = RANDOM_GEN.nextInt(Integer.MAX_VALUE);
      Path path =
          new Path(
              locksDir,
              String.format(LOCKS_DIR_TASK_PATTERN, getJobJtIdentifier(conf), taskIdCandidate));
      lockAcquired = tryCreateFile(conf, path);
    }

    return HadoopFormats.createTaskID(jobId, taskIdCandidate);
  }
@Test
  public void testTaskIdLockAcquire() {
    int tasksCount = 100;
    for (int i = 0; i < tasksCount; i++) {
      TaskID taskID = tested.acquireTaskIdLock(configuration);
      assertTrue(isFileExists(getTaskIdPath(taskID)));
    }

    String jobFolderName = getFileInJobFolder("");
    File jobFolder = new File(jobFolderName);
    assertTrue(jobFolder.isDirectory());
    // we have to multiply by 2 because crc files exists
    assertEquals(tasksCount * 2, jobFolder.list().length);
  }