@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); 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(eq(user), eq(conflictPath)); } finally { lfs.delete(base_path, true); if (dispatcher != null) { dispatcher.stop(); } } }
@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(); 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; } LOG.warn("Directory " + uniquePath + " already exists, " + "try next one."); if (delService != null) { delService.delete(getUser(), uniquePath); } } Path localPath = new Path(rPath, req.getPath().getName()); LocalizedResource rsrc = localrsrc.get(req); 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; }
LocalResourcesTrackerImpl implements LocalResourcesTracker { @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(); 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; } LOG.warn("Directory " + uniquePath + " already exists, " + "try next one."); if (delService != null) { delService.delete(getUser(), uniquePath); } } Path localPath = new Path(rPath, req.getPath().getName()); LocalizedResource rsrc = localrsrc.get(req); 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; } }
LocalResourcesTrackerImpl implements LocalResourcesTracker { @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(); 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; } LOG.warn("Directory " + uniquePath + " already exists, " + "try next one."); if (delService != null) { delService.delete(getUser(), uniquePath); } } Path localPath = new Path(rPath, req.getPath().getName()); LocalizedResource rsrc = localrsrc.get(req); 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; } LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher, boolean useLocalCacheDirectoryManager,
      Configuration conf, NMStateStoreService stateStore);   LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher,
      ConcurrentMap<LocalResourceRequest,LocalizedResource> localrsrc,
      boolean useLocalCacheDirectoryManager, Configuration conf,
      NMStateStoreService stateStore); }
LocalResourcesTrackerImpl implements LocalResourcesTracker { @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(); 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; } LOG.warn("Directory " + uniquePath + " already exists, " + "try next one."); if (delService != null) { delService.delete(getUser(), uniquePath); } } Path localPath = new Path(rPath, req.getPath().getName()); LocalizedResource rsrc = localrsrc.get(req); 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; } LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher, boolean useLocalCacheDirectoryManager,
      Configuration conf, NMStateStoreService stateStore);   LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher,
      ConcurrentMap<LocalResourceRequest,LocalizedResource> localrsrc,
      boolean useLocalCacheDirectoryManager, Configuration conf,
      NMStateStoreService stateStore); @Override synchronized void handle(ResourceEvent event); void incrementFileCountForLocalCacheDirectory(Path cacheDir); boolean isResourcePresent(LocalizedResource rsrc); @Override boolean remove(LocalizedResource rem, DeletionService delService); @Override String getUser(); @Override Iterator<LocalizedResource> iterator(); @Override Path getPathForLocalization(LocalResourceRequest req,
      Path localDirPath, DeletionService delService); @Override LocalizedResource getLocalizedResource(LocalResourceRequest request); }
LocalResourcesTrackerImpl implements LocalResourcesTracker { @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(); 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; } LOG.warn("Directory " + uniquePath + " already exists, " + "try next one."); if (delService != null) { delService.delete(getUser(), uniquePath); } } Path localPath = new Path(rPath, req.getPath().getName()); LocalizedResource rsrc = localrsrc.get(req); 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; } LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher, boolean useLocalCacheDirectoryManager,
      Configuration conf, NMStateStoreService stateStore);   LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher,
      ConcurrentMap<LocalResourceRequest,LocalizedResource> localrsrc,
      boolean useLocalCacheDirectoryManager, Configuration conf,
      NMStateStoreService stateStore); @Override synchronized void handle(ResourceEvent event); void incrementFileCountForLocalCacheDirectory(Path cacheDir); boolean isResourcePresent(LocalizedResource rsrc); @Override boolean remove(LocalizedResource rem, DeletionService delService); @Override String getUser(); @Override Iterator<LocalizedResource> iterator(); @Override Path getPathForLocalization(LocalResourceRequest req,
      Path localDirPath, DeletionService delService); @Override LocalizedResource getLocalizedResource(LocalResourceRequest request);  }