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); } HDFSSynchronization(String locksDir);   HDFSSynchronization(
      String locksDir, ThrowingFunction<Configuration, FileSystem, IOException> fileSystemFactory); @Override boolean tryAcquireJobLock(Configuration conf); @Override void releaseJobIdLock(Configuration conf); @Override TaskID acquireTaskIdLock(Configuration conf); @Override TaskAttemptID acquireTaskAttemptIdLock(Configuration conf, int taskId);  }
@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()))); } }