@SuppressWarnings("unchecked")
  @Override
  public void handle(ContainerAllocatorEvent event) {
    if (event.getType() == ContainerAllocator.EventType.CONTAINER_REQ) {
      LOG.info("Processing the event " + event.toString());
      // Assign the same container ID as the AM
      ContainerId cID =
          ContainerId.newContainerId(getContext().getApplicationAttemptId(),
            this.containerId.getContainerId());
      Container container = recordFactory.newRecordInstance(Container.class);
      container.setId(cID);
      NodeId nodeId = NodeId.newInstance(this.nmHost, this.nmPort);
      container.setResource(Resource.newInstance(0, 0));
      container.setNodeId(nodeId);
      container.setContainerToken(null);
      container.setNodeHttpAddress(this.nmHost + ":" + this.nmHttpPort);
      // send the container-assigned event to task attempt

      if (event.getAttemptID().getTaskId().getTaskType() == TaskType.MAP) {
        JobCounterUpdateEvent jce =
            new JobCounterUpdateEvent(event.getAttemptID().getTaskId()
                .getJobId());
        // TODO Setting OTHER_LOCAL_MAP for now.
        jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
        eventHandler.handle(jce);
      }
      eventHandler.handle(new TaskAttemptContainerAssignedEvent(
          event.getAttemptID(), container, applicationACLs));
    }
  }
@Test
  public void testAllocatedContainerResourceIsNotNull() {
    ArgumentCaptor<TaskAttemptContainerAssignedEvent> containerAssignedCaptor
        = ArgumentCaptor.forClass(TaskAttemptContainerAssignedEvent.class);
    @SuppressWarnings("unchecked")
    EventHandler<Event> eventHandler = mock(EventHandler.class);
    AppContext context = mock(AppContext.class) ;
    when(context.getEventHandler()).thenReturn(eventHandler);
    ContainerId containerId = ContainerId.fromString(
        "container_1427562107907_0002_01_000001");
    LocalContainerAllocator containerAllocator = new LocalContainerAllocator(
        mock(ClientService.class), context, "localhost", -1, -1, containerId);

    ContainerAllocatorEvent containerAllocatorEvent =
        createContainerRequestEvent();
    containerAllocator.handle(containerAllocatorEvent);

    verify(eventHandler, times(1)).handle(containerAssignedCaptor.capture());
    Container container = containerAssignedCaptor.getValue().getContainer();
    Resource containerResource = container.getResource();
    Assert.assertNotNull(containerResource);
    assertThat(containerResource.getMemorySize()).isEqualTo(0);
    assertThat(containerResource.getVirtualCores()).isEqualTo(0);
  }