@Test(timeout = 1000)
@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());
    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, 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, 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();
    }
  }
}