@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) {
        // The container sent a release event on a resource which 
        // 1) Failed
        // 2) Removed for some reason (ex. disk is no longer accessible)
        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:
      /*
       * If resource localization fails then Localized resource will be
       * removed from local cache.
       */
      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);

    // Remove the resource if its downloading and its reference count has
    // become 0 after RELEASE. This maybe because a container was killed while
    // localizing and no other container is referring to the resource.
    // NOTE: This should NOT be done for public resources since the
    //       download is not associated with a container-specific localizer.
    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");
      }
    }
  }
@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);

      // Registering event handlers.
      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);

      // Creating 2 containers for same application which will be requesting
      // same local resource.
      // Container 1 requesting local resource.
      ContainerId cId1 = BuilderUtils.newContainerId(1, 1, 1, 1);
      LocalizerContext lc1 = new LocalizerContext(user, cId1, null);
      ResourceEvent reqEvent1 =
          new ResourceRequestEvent(lr, LocalResourceVisibility.PRIVATE, lc1);

      // No resource request is initially present in local cache
      Assert.assertEquals(0, localrsrc.size());

      // Container-1 requesting local resource.
      tracker.handle(reqEvent1);
      dispatcher.await();

      // New localized Resource should have been added to local resource map
      // and the requesting container will be added to its waiting queue.
      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());

      // Container 2 requesting the resource
      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();

      // Container 2 should have been added to the waiting queue of the local
      // resource
      Assert.assertEquals(2, localrsrc.get(lr).getRefCount());
      Assert.assertTrue(localrsrc.get(lr).ref.contains(cId2));

      // Failing resource localization
      ResourceEvent resourceFailedEvent = new ResourceFailedLocalizationEvent(
          lr,(new Exception("test").getMessage()));
      
      // Backing up the resource to track its state change as it will be
      // removed after the failed event.
      LocalizedResource localizedResource = localrsrc.get(lr);
      
      tracker.handle(resourceFailedEvent);
      dispatcher.await();

      // After receiving failed resource event; all waiting containers will be
      // notified with Container Resource Failed Event.
      Assert.assertEquals(0, localrsrc.size());
      verify(containerEventHandler, timeout(1000).times(2)).handle(
        isA(ContainerResourceFailedEvent.class));
      Assert.assertEquals(ResourceState.FAILED, localizedResource.getState());

      // Container 1 trying to release the resource (This resource is already
      // deleted from the cache. This call should return silently without
      // exception.
      ResourceReleaseEvent relEvent1 = new ResourceReleaseEvent(lr, cId1);
      tracker.handle(relEvent1);
      dispatcher.await();

      // Container-3 now requests for the same resource. This request call
      // is coming prior to Container-2's release call.
      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();

      // Local resource cache now should have the requested resource and the
      // number of waiting containers should be 1.
      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));

      // Container-2 Releases the resource
      ResourceReleaseEvent relEvent2 = new ResourceReleaseEvent(lr, cId2);
      tracker.handle(relEvent2);
      dispatcher.await();

      // Making sure that there is no change in the cache after the release.
      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));
      
      // Sending ResourceLocalizedEvent to tracker. In turn resource should
      // send Container Resource Localized Event to waiting containers.
      Path localizedPath = new Path("/tmp/file1");
      ResourceLocalizedEvent localizedEvent =
          new ResourceLocalizedEvent(lr, localizedPath, 123L);
      tracker.handle(localizedEvent);
      dispatcher.await();
      
      // Verifying ContainerResourceLocalizedEvent .
      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());
      
      // Container-3 releasing the resource.
      ResourceReleaseEvent relEvent3 = new ResourceReleaseEvent(lr, cId3);
      tracker.handle(relEvent3);
      dispatcher.await();
      
      Assert.assertEquals(0, localrsrc.get(lr).getRefCount());
      
    } finally {
      if (dispatcher != null) {
        dispatcher.stop();
      }
    }
  }