@Override
  public Path getPathForLocalization(LocalResourceRequest req,
      Path localDirPath, DeletionService delService) {
    Path rPath = localDirPath;
    if (useLocalCacheDirectoryManager && localDirPath != null) {

      if (!directoryManagers.containsKey(localDirPath)) {
        directoryManagers.putIfAbsent(localDirPath,
          new LocalCacheDirectoryManager(conf));
      }
      LocalCacheDirectoryManager dir = directoryManagers.get(localDirPath);

      rPath = localDirPath;
      String hierarchicalPath = dir.getRelativePathForLocalization();
      // For most of the scenarios we will get root path only which
      // is an empty string
      if (!hierarchicalPath.isEmpty()) {
        rPath = new Path(localDirPath, hierarchicalPath);
      }
      inProgressLocalResourcesMap.put(req, rPath);
    }

    while (true) {
      Path uniquePath = new Path(rPath,
          Long.toString(uniqueNumberGenerator.incrementAndGet()));
      File file = new File(uniquePath.toUri().getRawPath());
      if (!file.exists()) {
        rPath = uniquePath;
        break;
      }
      // If the directory already exists, delete it and move to next one.
      LOG.warn("Directory " + uniquePath + " already exists, " +
          "try next one.");
      if (delService != null) {
        FileDeletionTask deletionTask = new FileDeletionTask(delService,
            getUser(), uniquePath, null);
        delService.delete(deletionTask);
      }
    }

    Path localPath = new Path(rPath, req.getPath().getName());
    LocalizedResource rsrc = localrsrc.get(req);
    if (rsrc == null) {
      LOG.warn("Resource " + req + " has been removed"
          + " and will no longer be localized");
      return null;
    }
    rsrc.setLocalPath(localPath);
    LocalResource lr = LocalResource.newInstance(req.getResource(),
        req.getType(), req.getVisibility(), req.getSize(),
        req.getTimestamp());
    try {
      stateStore.startResourceLocalization(user, appId,
          ((LocalResourcePBImpl) lr).getProto(), localPath);
    } catch (IOException e) {
      LOG.error("Unable to record localization start for " + rsrc, e);
    }
    return rPath;
  }
@Test
  @SuppressWarnings("unchecked")
  public void testGetPathForLocalization() throws Exception {
    FileContext lfs = FileContext.getLocalFSFileContext();
    Path base_path = new Path("target",
        TestLocalResourcesTrackerImpl.class.getSimpleName());
    final String user = "someuser";
    final ApplicationId appId = ApplicationId.newInstance(1, 1);
    Configuration conf = new YarnConfiguration();
    DrainDispatcher dispatcher = null;
    dispatcher = createDispatcher(conf);
    EventHandler<LocalizerEvent> localizerEventHandler =
        mock(EventHandler.class);
    EventHandler<LocalizerEvent> containerEventHandler =
        mock(EventHandler.class);
    dispatcher.register(LocalizerEventType.class, localizerEventHandler);
    dispatcher.register(ContainerEventType.class, containerEventHandler);
    NMStateStoreService stateStore = mock(NMStateStoreService.class);
    DeletionService delService = mock(DeletionService.class);
    try {
      LocalResourceRequest req1 = createLocalResourceRequest(user, 1, 1,
          LocalResourceVisibility.PUBLIC);
      LocalizedResource lr1 = createLocalizedResource(req1, dispatcher);
      ConcurrentMap<LocalResourceRequest, LocalizedResource> localrsrc =
          new ConcurrentHashMap<LocalResourceRequest, LocalizedResource>();
      localrsrc.put(req1, lr1);
      LocalResourcesTrackerImpl tracker = new LocalResourcesTrackerImpl(user,
          appId, dispatcher, localrsrc, true, conf, stateStore, null);
      Path conflictPath = new Path(base_path, "10");
      Path qualifiedConflictPath = lfs.makeQualified(conflictPath);
      lfs.mkdir(qualifiedConflictPath, null, true);
      Path rPath = tracker.getPathForLocalization(req1, base_path,
          delService);
      Assert.assertFalse(lfs.util().exists(rPath));
      verify(delService, times(1)).delete(argThat(new FileDeletionMatcher(
          delService, user, conflictPath, null)));
    } finally {
      lfs.delete(base_path, true);
      if (dispatcher != null) {
        dispatcher.stop();
      }
    }
  }