LocalResourcesTrackerImpl implements LocalResourcesTracker { @Override public synchronized void handle(ResourceEvent event) { LocalResourceRequest req = event.getLocalResourceRequest(); LocalizedResource rsrc = localrsrc.get(req); switch (event.getType()) { case LOCALIZED: if (useLocalCacheDirectoryManager) { inProgressLocalResourcesMap.remove(req); } break; case REQUEST: if (rsrc != null && (!isResourcePresent(rsrc))) { LOG.info("Resource " + rsrc.getLocalPath() + " is missing, localizing it again"); removeResource(req); rsrc = null; } if (null == rsrc) { rsrc = new LocalizedResource(req, dispatcher); localrsrc.put(req, rsrc); } break; case RELEASE: if (null == rsrc) { ResourceReleaseEvent relEvent = (ResourceReleaseEvent) event; LOG.info("Container " + relEvent.getContainer() + " sent RELEASE event on a resource request " + req + " not present in cache."); return; } break; case LOCALIZATION_FAILED: removeResource(req); break; case RECOVERED: if (rsrc != null) { LOG.warn("Ignoring attempt to recover existing resource " + rsrc); return; } rsrc = recoverResource(req, (ResourceRecoveredEvent) event); localrsrc.put(req, rsrc); break; } if (rsrc == null) { LOG.warn("Received " + event.getType() + " event for request " + req + " but localized resource is missing"); return; } rsrc.handle(event); if (event.getType() == ResourceEventType.RELEASE) { if (rsrc.getState() == ResourceState.DOWNLOADING && rsrc.getRefCount() <= 0 && rsrc.getRequest().getVisibility() != LocalResourceVisibility.PUBLIC) { removeResource(req); } } if (event.getType() == ResourceEventType.LOCALIZED) { if (rsrc.getLocalPath() != null) { try { stateStore.finishResourceLocalization(user, appId, buildLocalizedResourceProto(rsrc)); } catch (IOException ioe) { LOG.error("Error storing resource state for " + rsrc, ioe); } } else { LOG.warn("Resource " + rsrc + " localized without a location"); } } } LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher, boolean useLocalCacheDirectoryManager,
      Configuration conf, NMStateStoreService stateStore,
      LocalDirsHandlerService dirHandler);   LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher,
      ConcurrentMap<LocalResourceRequest, LocalizedResource> localrsrc,
      boolean useLocalCacheDirectoryManager, Configuration conf,
      NMStateStoreService stateStore, LocalDirsHandlerService dirHandler); @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);  }
@Test(timeout = 10000) @SuppressWarnings("unchecked") public void testLocalResourceCache() { String user = "testuser"; DrainDispatcher dispatcher = null; try { Configuration conf = new Configuration(); dispatcher = createDispatcher(conf); EventHandler<LocalizerEvent> localizerEventHandler = mock(EventHandler.class); EventHandler<ContainerEvent> containerEventHandler = mock(EventHandler.class); dispatcher.register(LocalizerEventType.class, localizerEventHandler); dispatcher.register(ContainerEventType.class, containerEventHandler); ConcurrentMap<LocalResourceRequest, LocalizedResource> localrsrc = new ConcurrentHashMap<LocalResourceRequest, LocalizedResource>(); LocalResourcesTracker tracker = new LocalResourcesTrackerImpl(user, null, dispatcher, localrsrc, true, conf, new NMNullStateStoreService(), null); LocalResourceRequest lr = createLocalResourceRequest(user, 1, 1, LocalResourceVisibility.PUBLIC); ContainerId cId1 = BuilderUtils.newContainerId(1, 1, 1, 1); LocalizerContext lc1 = new LocalizerContext(user, cId1, null); ResourceEvent reqEvent1 = new ResourceRequestEvent(lr, LocalResourceVisibility.PRIVATE, lc1); Assert.assertEquals(0, localrsrc.size()); tracker.handle(reqEvent1); dispatcher.await(); Assert.assertEquals(1, localrsrc.size()); Assert.assertTrue(localrsrc.containsKey(lr)); Assert.assertEquals(1, localrsrc.get(lr).getRefCount()); Assert.assertTrue(localrsrc.get(lr).ref.contains(cId1)); Assert.assertEquals(ResourceState.DOWNLOADING, localrsrc.get(lr) .getState()); ContainerId cId2 = BuilderUtils.newContainerId(1, 1, 1, 2); LocalizerContext lc2 = new LocalizerContext(user, cId2, null); ResourceEvent reqEvent2 = new ResourceRequestEvent(lr, LocalResourceVisibility.PRIVATE, lc2); tracker.handle(reqEvent2); dispatcher.await(); Assert.assertEquals(2, localrsrc.get(lr).getRefCount()); Assert.assertTrue(localrsrc.get(lr).ref.contains(cId2)); ResourceEvent resourceFailedEvent = new ResourceFailedLocalizationEvent( lr,(new Exception("test").getMessage())); LocalizedResource localizedResource = localrsrc.get(lr); tracker.handle(resourceFailedEvent); dispatcher.await(); Assert.assertEquals(0, localrsrc.size()); verify(containerEventHandler, timeout(1000).times(2)).handle( isA(ContainerResourceFailedEvent.class)); Assert.assertEquals(ResourceState.FAILED, localizedResource.getState()); ResourceReleaseEvent relEvent1 = new ResourceReleaseEvent(lr, cId1); tracker.handle(relEvent1); dispatcher.await(); ContainerId cId3 = BuilderUtils.newContainerId(1, 1, 1, 3); LocalizerContext lc3 = new LocalizerContext(user, cId3, null); ResourceEvent reqEvent3 = new ResourceRequestEvent(lr, LocalResourceVisibility.PRIVATE, lc3); tracker.handle(reqEvent3); dispatcher.await(); Assert.assertEquals(1, localrsrc.size()); Assert.assertTrue(localrsrc.containsKey(lr)); Assert.assertEquals(1, localrsrc.get(lr).getRefCount()); Assert.assertTrue(localrsrc.get(lr).ref.contains(cId3)); ResourceReleaseEvent relEvent2 = new ResourceReleaseEvent(lr, cId2); tracker.handle(relEvent2); dispatcher.await(); Assert.assertEquals(1, localrsrc.size()); Assert.assertTrue(localrsrc.containsKey(lr)); Assert.assertEquals(1, localrsrc.get(lr).getRefCount()); Assert.assertTrue(localrsrc.get(lr).ref.contains(cId3)); Path localizedPath = new Path("/tmp/file1"); ResourceLocalizedEvent localizedEvent = new ResourceLocalizedEvent(lr, localizedPath, 123L); tracker.handle(localizedEvent); dispatcher.await(); verify(containerEventHandler, timeout(1000).times(1)).handle( isA(ContainerResourceLocalizedEvent.class)); Assert.assertEquals(ResourceState.LOCALIZED, localrsrc.get(lr) .getState()); Assert.assertEquals(1, localrsrc.get(lr).getRefCount()); ResourceReleaseEvent relEvent3 = new ResourceReleaseEvent(lr, cId3); tracker.handle(relEvent3); dispatcher.await(); Assert.assertEquals(0, localrsrc.get(lr).getRefCount()); } finally { if (dispatcher != null) { dispatcher.stop(); } } }
@Test @SuppressWarnings("unchecked") public void testReleaseWhileDownloading() throws Exception { String user = "testuser"; DrainDispatcher dispatcher = null; try { Configuration conf = new Configuration(); 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); ContainerId cId = BuilderUtils.newContainerId(1, 1, 1, 1); LocalizerContext lc = new LocalizerContext(user, cId, null); LocalResourceRequest req = createLocalResourceRequest(user, 1, 1, LocalResourceVisibility.PUBLIC); LocalizedResource lr = createLocalizedResource(req, dispatcher); ConcurrentMap<LocalResourceRequest, LocalizedResource> localrsrc = new ConcurrentHashMap<LocalResourceRequest, LocalizedResource>(); localrsrc.put(req, lr); LocalResourcesTracker tracker = new LocalResourcesTrackerImpl(user, null, dispatcher, localrsrc, false, conf, new NMNullStateStoreService(), null); ResourceEvent reqEvent = new ResourceRequestEvent(req, LocalResourceVisibility.PUBLIC, lc); tracker.handle(reqEvent); ResourceEvent relEvent = new ResourceReleaseEvent(req, cId); tracker.handle(relEvent); ResourceLocalizedEvent rle = new ResourceLocalizedEvent(req, new Path("file: tracker.handle(rle); dispatcher.await(); } finally { if (dispatcher != null) { dispatcher.stop(); } } }