@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()); 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(); } } }
@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 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 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);   LocalResourcesTrackerImpl(String user, ApplicationId appId,
      Dispatcher dispatcher,
      ConcurrentMap<LocalResourceRequest,LocalizedResource> localrsrc,
      boolean useLocalCacheDirectoryManager, Configuration conf,
      NMStateStoreService stateStore); }
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);   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 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);   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);  }