import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
public void handle(ContainerEvent event) {
  if (event instanceof ContainerExitEvent) {
    containerExitEventOccured = true;
    ContainerExitEvent exitEvent = (ContainerExitEvent) event;
    Assert.assertEquals(ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
        exitEvent.getType());
    LOG.info("Diagnostic Info : " + exitEvent.getDiagnosticInfo());
    if (testForMultiFile) {
      Assert.assertTrue("Should contain the Multi file information",
          exitEvent.getDiagnosticInfo().contains("Error files : "));
    }
    Assert.assertTrue(
        "Should contain the error Log message with tail size info",
        exitEvent.getDiagnosticInfo()
            .contains("Last "
                + YarnConfiguration.DEFAULT_NM_CONTAINER_STDERR_BYTES
                + " bytes of"));
    Assert.assertTrue("Should contain contents of error Log",
        exitEvent.getDiagnosticInfo().contains(
            INVALID_JAVA_HOME + "/bin/java: No such file or directory"));
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
@Test (timeout = 10000)
public void testCallFailureWithNullLocalizedResources() {
  Container container = mock(Container.class);
  when(container.getContainerId()).thenReturn(ContainerId.newContainerId(
      ApplicationAttemptId.newInstance(ApplicationId.newInstance(
          System.currentTimeMillis(), 1), 1), 1));
  ContainerLaunchContext clc = mock(ContainerLaunchContext.class);
  when(clc.getCommands()).thenReturn(Collections.<String>emptyList());
  when(container.getLaunchContext()).thenReturn(clc);
  when(container.getLocalizedResources()).thenReturn(null);
  Dispatcher dispatcher = mock(Dispatcher.class);
  EventHandler eventHandler = new EventHandler() {
    public void handle(Event event) {
      Assert.assertTrue(event instanceof ContainerExitEvent);
      ContainerExitEvent exitEvent = (ContainerExitEvent) event;
      Assert.assertEquals(ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
          exitEvent.getType());
    }
  };
  when(dispatcher.getEventHandler()).thenReturn(eventHandler);
  ContainerLaunch launch = new ContainerLaunch(context, new Configuration(),
      dispatcher, exec, null, container, dirsHandler, containerManager);
  launch.call();
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
protected ContainersLauncher createContainersLauncher(Context context,
    ContainerExecutor exec) {
  return new ContainersLauncher(context, super.dispatcher, exec,
                                super.dirsHandler, this) {
    @Override
    public void handle(ContainersLauncherEvent event) {
      Container container = event.getContainer();
      ContainerId containerId = container.getContainerId();
      switch (event.getType()) {
      case LAUNCH_CONTAINER:
        dispatcher.getEventHandler().handle(
            new ContainerEvent(containerId,
                ContainerEventType.CONTAINER_LAUNCHED));
        break;
      case CLEANUP_CONTAINER:
        dispatcher.getEventHandler().handle(
            new ContainerExitEvent(containerId,
                ContainerEventType.CONTAINER_KILLED_ON_REQUEST, 0,
                "Container exited with exit code 0."));
        break;
      }
    }
  };
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
public void handle(ContainerEvent event) {
  if (event instanceof ContainerExitEvent) {
    containerExitEventOccured = true;
    ContainerExitEvent exitEvent = (ContainerExitEvent) event;
    Assert.assertEquals(ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
        exitEvent.getType());
    LOG.info("Diagnostic Info : " + exitEvent.getDiagnosticInfo());
    if (testForMultiFile) {
      Assert.assertTrue("Should contain the Multi file information",
          exitEvent.getDiagnosticInfo().contains("Error files: "));
    }
    Assert.assertTrue(
        "Should contain the error Log message with tail size info",
        exitEvent.getDiagnosticInfo()
            .contains("Last "
                + YarnConfiguration.DEFAULT_NM_CONTAINER_STDERR_BYTES
                + " bytes of"));
    Assert.assertTrue("Should contain contents of error Log",
        exitEvent.getDiagnosticInfo().contains(
            INVALID_JAVA_HOME + "/bin/java: No such file or directory"));
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Before
public void setup() {
  executor = new MockExecutor();
  dispatcher = new AsyncDispatcher();
  context = Mockito.mock(Context.class);
  Mockito.doReturn(new ConcurrentSkipListMap<ContainerId, Container>())
      .when(context).getContainers();
  conf = new Configuration();
  conf.set(
      YarnConfiguration.NM_CONTAINER_MON_RESOURCE_CALCULATOR,
      MockResourceCalculatorPlugin.class.getCanonicalName());
  conf.set(
      YarnConfiguration.NM_CONTAINER_MON_PROCESS_TREE,
      MockResourceCalculatorProcessTree.class.getCanonicalName());
  dispatcher.init(conf);
  dispatcher.start();
  containerEventHandler = new MockContainerEventHandler();
  dispatcher.register(ContainerEventType.class, containerEventHandler);
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
protected ContainersLauncher createContainersLauncher(Context context,
    ContainerExecutor exec) {
  return new ContainersLauncher(context, super.dispatcher, exec,
                                super.dirsHandler) {
    @Override
    public void handle(ContainersLauncherEvent event) {
      Container container = event.getContainer();
      ContainerId containerId = container.getContainerId();
      switch (event.getType()) {
      case LAUNCH_CONTAINER:
        dispatcher.getEventHandler().handle(
            new ContainerEvent(containerId,
                ContainerEventType.CONTAINER_LAUNCHED));
        break;
      case CLEANUP_CONTAINER:
        dispatcher.getEventHandler().handle(
            new ContainerExitEvent(containerId,
                ContainerEventType.CONTAINER_KILLED_ON_REQUEST, 0,
                "Container exited with exit code 0."));
        break;
      }
    }
  };
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
@Test
public void testCallFailureWithNullLocalizedResources() {
  Container container = mock(Container.class);
  when(container.getContainerId()).thenReturn(ContainerId.newInstance(
      ApplicationAttemptId.newInstance(ApplicationId.newInstance(
          System.currentTimeMillis(), 1), 1), 1));
  ContainerLaunchContext clc = mock(ContainerLaunchContext.class);
  when(clc.getCommands()).thenReturn(Collections.<String>emptyList());
  when(container.getLaunchContext()).thenReturn(clc);
  when(container.getLocalizedResources()).thenReturn(null);
  Dispatcher dispatcher = mock(Dispatcher.class);
  EventHandler eventHandler = new EventHandler() {
    public void handle(Event event) {
      Assert.assertTrue(event instanceof ContainerExitEvent);
      ContainerExitEvent exitEvent = (ContainerExitEvent) event;
      Assert.assertEquals(ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
          exitEvent.getType());
    }
  };
  when(dispatcher.getEventHandler()).thenReturn(eventHandler);
  ContainerLaunch launch = new ContainerLaunch(context, new Configuration(),
      dispatcher, exec, null, container, dirsHandler, containerManager);
  launch.call();
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Override
public void handle(ContainerEvent event) {
  if (event.getType() == ContainerEventType.KILL_CONTAINER) {
    synchronized (killedContainer) {
      killedContainer.add(event.getContainerID());
    }
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private void handleCleanupContainerResources(
    ContainerLocalizationCleanupEvent rsrcCleanup) {
  Container c = rsrcCleanup.getContainer();
  Map<LocalResourceVisibility, Collection<LocalResourceRequest>> rsrcs =
    rsrcCleanup.getResources();
  for (Map.Entry<LocalResourceVisibility, Collection<LocalResourceRequest>> e :
       rsrcs.entrySet()) {
    LocalResourcesTracker tracker = getLocalResourcesTracker(e.getKey(), c.getUser(), 
        c.getContainerId().getApplicationAttemptId()
        .getApplicationId());
    for (LocalResourceRequest req : e.getValue()) {
      tracker.handle(new ResourceReleaseEvent(req,
          c.getContainerId()));
    }
  }
  String locId = ConverterUtils.toString(c.getContainerId());
  localizerTracker.cleanupPrivLocalizers(locId);
  
  // Delete the container directories
  String userName = c.getUser();
  String containerIDStr = c.toString();
  String appIDStr = ConverterUtils.toString(
      c.getContainerId().getApplicationAttemptId().getApplicationId());
  
  // Try deleting from good local dirs and full local dirs because a dir might
  // have gone bad while the app was running(disk full). In addition
  // a dir might have become good while the app was running.
  // Check if the container dir exists and if it does, try to delete it

  for (String localDir : dirsHandler.getLocalDirsForCleanup()) {
    // Delete the user-owned container-dir
    Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE);
    Path userdir = new Path(usersdir, userName);
    Path allAppsdir = new Path(userdir, ContainerLocalizer.APPCACHE);
    Path appDir = new Path(allAppsdir, appIDStr);
    Path containerDir = new Path(appDir, containerIDStr);
    submitDirForDeletion(userName, containerDir);

    // Delete the nmPrivate container-dir

    Path sysDir = new Path(localDir, NM_PRIVATE_DIR);
    Path appSysDir = new Path(sysDir, appIDStr);
    Path containerSysDir = new Path(appSysDir, containerIDStr);
    submitDirForDeletion(null, containerSysDir);
  }

  dispatcher.getEventHandler().handle(
      new ContainerEvent(c.getContainerId(),
          ContainerEventType.CONTAINER_RESOURCES_CLEANEDUP));
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
/**
 * Tries to tail and fetch TAIL_SIZE_IN_BYTES of data from the error log.
 * ErrorLog filename is not fixed and depends upon app, hence file name
 * pattern is used.
 * @param containerID
 * @param ret
 * @param containerLogDir
 * @param diagnosticInfo
 */
@SuppressWarnings("unchecked")
private void handleContainerExitWithFailure(ContainerId containerID, int ret,
    Path containerLogDir, StringBuilder diagnosticInfo) {
  LOG.warn(diagnosticInfo);

  String errorFileNamePattern =
      conf.get(YarnConfiguration.NM_CONTAINER_STDERR_PATTERN,
          YarnConfiguration.DEFAULT_NM_CONTAINER_STDERR_PATTERN);
  FSDataInputStream errorFileIS = null;
  try {
    FileSystem fileSystem = FileSystem.getLocal(conf).getRaw();
    FileStatus[] errorFileStatuses = fileSystem
        .globStatus(new Path(containerLogDir, errorFileNamePattern));
    if (errorFileStatuses != null && errorFileStatuses.length != 0) {
      long tailSizeInBytes =
          conf.getLong(YarnConfiguration.NM_CONTAINER_STDERR_BYTES,
              YarnConfiguration.DEFAULT_NM_CONTAINER_STDERR_BYTES);
      Path errorFile = errorFileStatuses[0].getPath();
      long fileSize = errorFileStatuses[0].getLen();

      // if more than one file matches the stderr pattern, take the latest
      // modified file, and also append the file names in the diagnosticInfo
      if (errorFileStatuses.length > 1) {
        String[] errorFileNames = new String[errorFileStatuses.length];
        long latestModifiedTime = errorFileStatuses[0].getModificationTime();
        errorFileNames[0] = errorFileStatuses[0].getPath().getName();
        for (int i = 1; i < errorFileStatuses.length; i++) {
          errorFileNames[i] = errorFileStatuses[i].getPath().getName();
          if (errorFileStatuses[i]
              .getModificationTime() > latestModifiedTime) {
            latestModifiedTime = errorFileStatuses[i].getModificationTime();
            errorFile = errorFileStatuses[i].getPath();
            fileSize = errorFileStatuses[i].getLen();
          }
        }
        diagnosticInfo.append("Error files : ")
            .append(StringUtils.join(", ", errorFileNames)).append(".\n");
      }

      long startPosition =
          (fileSize < tailSizeInBytes) ? 0 : fileSize - tailSizeInBytes;
      int bufferSize =
          (int) ((fileSize < tailSizeInBytes) ? fileSize : tailSizeInBytes);
      byte[] tailBuffer = new byte[bufferSize];
      errorFileIS = fileSystem.open(errorFile);
      errorFileIS.readFully(startPosition, tailBuffer);

      diagnosticInfo.append("Last ").append(tailSizeInBytes)
          .append(" bytes of ").append(errorFile.getName()).append(" :\n")
          .append(new String(tailBuffer, StandardCharsets.UTF_8));
    }
  } catch (IOException e) {
    LOG.error("Failed to get tail of the container's error log file", e);
  } finally {
    IOUtils.cleanup(LOG, errorFileIS);
  }

  this.dispatcher.getEventHandler()
      .handle(new ContainerExitEvent(containerID,
          ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret,
          diagnosticInfo.toString()));
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
public ContainerManagerImpl(Context context, ContainerExecutor exec,
    DeletionService deletionContext, NodeStatusUpdater nodeStatusUpdater,
    NodeManagerMetrics metrics, ApplicationACLsManager aclsManager,
    LocalDirsHandlerService dirsHandler) {
  super(ContainerManagerImpl.class.getName());
  this.context = context;
  this.dirsHandler = dirsHandler;

  // ContainerManager level dispatcher.
  dispatcher = new AsyncDispatcher();
  this.deletionService = deletionContext;
  this.metrics = metrics;

  rsrcLocalizationSrvc =
      createResourceLocalizationService(exec, deletionContext, context);
  addService(rsrcLocalizationSrvc);

  containersLauncher = createContainersLauncher(context, exec);
  addService(containersLauncher);

  this.nodeStatusUpdater = nodeStatusUpdater;
  this.aclsManager = aclsManager;

  // Start configurable services
  auxiliaryServices = new AuxServices();
  auxiliaryServices.registerServiceListener(this);
  addService(auxiliaryServices);

  this.containersMonitor =
      new ContainersMonitorImpl(exec, dispatcher, this.context);
  addService(this.containersMonitor);

  dispatcher.register(ContainerEventType.class,
      new ContainerEventDispatcher());
  dispatcher.register(ApplicationEventType.class,
      new ApplicationEventDispatcher());
  dispatcher.register(LocalizationEventType.class, rsrcLocalizationSrvc);
  dispatcher.register(AuxServicesEventType.class, auxiliaryServices);
  dispatcher.register(ContainersMonitorEventType.class, containersMonitor);
  dispatcher.register(ContainersLauncherEventType.class, containersLauncher);
  
  addService(dispatcher);

  ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  this.readLock = lock.readLock();
  this.writeLock = lock.writeLock();
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Test(timeout=10000)
@SuppressWarnings("unchecked")
public void testConsistency() {
  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 cId1 = BuilderUtils.newContainerId(1, 1, 1, 1);
    LocalizerContext lc1 = new LocalizerContext(user, cId1, null);
    LocalResourceRequest req1 = createLocalResourceRequest(user, 1, 1,
        LocalResourceVisibility.PUBLIC);
    LocalizedResource lr1 = createLocalizedResource(req1, dispatcher);
    ConcurrentMap<LocalResourceRequest, LocalizedResource> localrsrc = new ConcurrentHashMap<LocalResourceRequest, LocalizedResource>();
    localrsrc.put(req1, lr1);
    LocalResourcesTracker tracker = new LocalResourcesTrackerImpl(user,
        null, dispatcher, localrsrc, false, conf,
        new NMNullStateStoreService(), null);

    ResourceEvent req11Event = new ResourceRequestEvent(req1,
        LocalResourceVisibility.PUBLIC, lc1);

    ResourceEvent rel11Event = new ResourceReleaseEvent(req1, cId1);

    // Localize R1 for C1
    tracker.handle(req11Event);

    dispatcher.await();

    // Verify refCount for R1 is 1
    Assert.assertEquals(1, lr1.getRefCount());

    dispatcher.await();
    verifyTrackedResourceCount(tracker, 1);

    // Localize resource1
    ResourceLocalizedEvent rle = new ResourceLocalizedEvent(req1, new Path(
        "file:///tmp/r1"), 1);
    lr1.handle(rle);
    Assert.assertTrue(lr1.getState().equals(ResourceState.LOCALIZED));
    Assert.assertTrue(createdummylocalizefile(new Path("file:///tmp/r1")));
    LocalizedResource rsrcbefore = tracker.iterator().next();
    File resFile = new File(lr1.getLocalPath().toUri().getRawPath()
        .toString());
    Assert.assertTrue(resFile.exists());
    Assert.assertTrue(resFile.delete());

    // Localize R1 for C1
    tracker.handle(req11Event);

    dispatcher.await();
    lr1.handle(rle);
    Assert.assertTrue(lr1.getState().equals(ResourceState.LOCALIZED));
    LocalizedResource rsrcafter = tracker.iterator().next();
    if (rsrcbefore == rsrcafter) {
      Assert.fail("Localized resource should not be equal");
    }
    // Release resource1
    tracker.handle(rel11Event);
  } finally {
    if (dispatcher != null) {
      dispatcher.stop();
    }
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testStateStoreSuccessfulLocalization() throws Exception {
  final String user = "someuser";
  final ApplicationId appId = ApplicationId.newInstance(1, 1);
  // This is a random path. NO File creation will take place at this place.
  final Path localDir = new Path("/tmp");
  Configuration conf = new YarnConfiguration();
  DrainDispatcher dispatcher = null;
  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);
  DeletionService mockDelService = mock(DeletionService.class);
  NMStateStoreService stateStore = mock(NMStateStoreService.class);

  try {
    LocalResourcesTracker tracker = new LocalResourcesTrackerImpl(user,
        appId, dispatcher, false, conf, stateStore);
    // Container 1 needs lr1 resource
    ContainerId cId1 = BuilderUtils.newContainerId(1, 1, 1, 1);
    LocalResourceRequest lr1 = createLocalResourceRequest(user, 1, 1,
        LocalResourceVisibility.APPLICATION);
    LocalizerContext lc1 = new LocalizerContext(user, cId1, null);

    // Container 1 requests lr1 to be localized
    ResourceEvent reqEvent1 = new ResourceRequestEvent(lr1,
        LocalResourceVisibility.APPLICATION, lc1);
    tracker.handle(reqEvent1);
    dispatcher.await();

    // Simulate the process of localization of lr1
    Path hierarchicalPath1 = tracker.getPathForLocalization(lr1, localDir,
        null);

    ArgumentCaptor<LocalResourceProto> localResourceCaptor =
        ArgumentCaptor.forClass(LocalResourceProto.class);
    ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class);
    verify(stateStore).startResourceLocalization(eq(user), eq(appId),
        localResourceCaptor.capture(), pathCaptor.capture());
    LocalResourceProto lrProto = localResourceCaptor.getValue();
    Path localizedPath1 = pathCaptor.getValue();
    Assert.assertEquals(lr1,
        new LocalResourceRequest(new LocalResourcePBImpl(lrProto)));
    Assert.assertEquals(hierarchicalPath1, localizedPath1.getParent());

    // Simulate lr1 getting localized
    ResourceLocalizedEvent rle1 =
        new ResourceLocalizedEvent(lr1, pathCaptor.getValue(), 120);
    tracker.handle(rle1);
    dispatcher.await();

    ArgumentCaptor<LocalizedResourceProto> localizedProtoCaptor =
        ArgumentCaptor.forClass(LocalizedResourceProto.class);
    verify(stateStore).finishResourceLocalization(eq(user), eq(appId),
        localizedProtoCaptor.capture());
    LocalizedResourceProto localizedProto = localizedProtoCaptor.getValue();
    Assert.assertEquals(lr1, new LocalResourceRequest(
        new LocalResourcePBImpl(localizedProto.getResource())));
    Assert.assertEquals(localizedPath1.toString(),
        localizedProto.getLocalPath());
    LocalizedResource localizedRsrc1 = tracker.getLocalizedResource(lr1);
    Assert.assertNotNull(localizedRsrc1);

    // simulate release and retention processing
    tracker.handle(new ResourceReleaseEvent(lr1, cId1));
    dispatcher.await();
    boolean removeResult = tracker.remove(localizedRsrc1, mockDelService);

    Assert.assertTrue(removeResult);
    verify(stateStore).removeLocalizedResource(eq(user), eq(appId),
        eq(localizedPath1));
  } finally {
    if (dispatcher != null) {
      dispatcher.stop();
    }
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testStateStoreFailedLocalization() throws Exception {
  final String user = "someuser";
  final ApplicationId appId = ApplicationId.newInstance(1, 1);
  // This is a random path. NO File creation will take place at this place.
  final Path localDir = new Path("/tmp");
  Configuration conf = new YarnConfiguration();
  DrainDispatcher dispatcher = null;
  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);
  NMStateStoreService stateStore = mock(NMStateStoreService.class);

  try {
    LocalResourcesTracker tracker = new LocalResourcesTrackerImpl(user,
        appId, dispatcher, false, conf, stateStore);
    // Container 1 needs lr1 resource
    ContainerId cId1 = BuilderUtils.newContainerId(1, 1, 1, 1);
    LocalResourceRequest lr1 = createLocalResourceRequest(user, 1, 1,
        LocalResourceVisibility.APPLICATION);
    LocalizerContext lc1 = new LocalizerContext(user, cId1, null);

    // Container 1 requests lr1 to be localized
    ResourceEvent reqEvent1 = new ResourceRequestEvent(lr1,
        LocalResourceVisibility.APPLICATION, lc1);
    tracker.handle(reqEvent1);
    dispatcher.await();

    // Simulate the process of localization of lr1
    Path hierarchicalPath1 = tracker.getPathForLocalization(lr1, localDir,
        null);

    ArgumentCaptor<LocalResourceProto> localResourceCaptor =
        ArgumentCaptor.forClass(LocalResourceProto.class);
    ArgumentCaptor<Path> pathCaptor = ArgumentCaptor.forClass(Path.class);
    verify(stateStore).startResourceLocalization(eq(user), eq(appId),
        localResourceCaptor.capture(), pathCaptor.capture());
    LocalResourceProto lrProto = localResourceCaptor.getValue();
    Path localizedPath1 = pathCaptor.getValue();
    Assert.assertEquals(lr1,
        new LocalResourceRequest(new LocalResourcePBImpl(lrProto)));
    Assert.assertEquals(hierarchicalPath1, localizedPath1.getParent());

    ResourceFailedLocalizationEvent rfe1 =
        new ResourceFailedLocalizationEvent(
            lr1, new Exception("Test").toString());
    tracker.handle(rfe1);
    dispatcher.await();
    verify(stateStore).removeLocalizedResource(eq(user), eq(appId),
        eq(localizedPath1));
  } finally {
    if (dispatcher != null) {
      dispatcher.stop();
    }
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testRecoveredResource() throws Exception {
  final String user = "someuser";
  final ApplicationId appId = ApplicationId.newInstance(1, 1);
  // This is a random path. NO File creation will take place at this place.
  final Path localDir = new Path("/tmp/localdir");
  Configuration conf = new YarnConfiguration();
  DrainDispatcher dispatcher = null;
  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);
  NMStateStoreService stateStore = mock(NMStateStoreService.class);

  try {
    LocalResourcesTracker tracker = new LocalResourcesTrackerImpl(user,
        appId, dispatcher, false, conf, stateStore);
    // Container 1 needs lr1 resource
    ContainerId cId1 = BuilderUtils.newContainerId(1, 1, 1, 1);
    LocalResourceRequest lr1 = createLocalResourceRequest(user, 1, 1,
        LocalResourceVisibility.APPLICATION);
    Assert.assertNull(tracker.getLocalizedResource(lr1));
    final long localizedId1 = 52;
    Path hierarchicalPath1 = new Path(localDir,
        Long.toString(localizedId1));
    Path localizedPath1 = new Path(hierarchicalPath1, "resource.jar");
    tracker.handle(new ResourceRecoveredEvent(lr1, localizedPath1, 120));
    dispatcher.await();
    Assert.assertNotNull(tracker.getLocalizedResource(lr1));

    // verify new paths reflect recovery of previous resources
    LocalResourceRequest lr2 = createLocalResourceRequest(user, 2, 2,
        LocalResourceVisibility.APPLICATION);
    LocalizerContext lc2 = new LocalizerContext(user, cId1, null);
    ResourceEvent reqEvent2 = new ResourceRequestEvent(lr2,
        LocalResourceVisibility.APPLICATION, lc2);
    tracker.handle(reqEvent2);
    dispatcher.await();
    Path hierarchicalPath2 = tracker.getPathForLocalization(lr2, localDir,
        null);
    long localizedId2 = Long.parseLong(hierarchicalPath2.getName());
    Assert.assertEquals(localizedId1 + 1, localizedId2);
  } finally {
    if (dispatcher != null) {
      dispatcher.stop();
    }
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testGetPathForLocalization() throws Exception {
  FileContext lfs = FileContext.getLocalFSFileContext();
  Path base_path = new Path("target",
      TestLocalResourcesTrackerImpl.class.getSimpleName());
  final String user = "someuser";
  final ApplicationId appId = ApplicationId.newInstance(1, 1);
  Configuration conf = new YarnConfiguration();
  DrainDispatcher dispatcher = null;
  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);
  NMStateStoreService stateStore = mock(NMStateStoreService.class);
  DeletionService delService = mock(DeletionService.class);
  try {
    LocalResourceRequest req1 = createLocalResourceRequest(user, 1, 1,
        LocalResourceVisibility.PUBLIC);
    LocalizedResource lr1 = createLocalizedResource(req1, dispatcher);
    ConcurrentMap<LocalResourceRequest, LocalizedResource> localrsrc =
        new ConcurrentHashMap<LocalResourceRequest, LocalizedResource>();
    localrsrc.put(req1, lr1);
    LocalResourcesTrackerImpl tracker = new LocalResourcesTrackerImpl(user,
        appId, dispatcher, localrsrc, true, conf, stateStore, null);
    Path conflictPath = new Path(base_path, "10");
    Path qualifiedConflictPath = lfs.makeQualified(conflictPath);
    lfs.mkdir(qualifiedConflictPath, null, true);
    Path rPath = tracker.getPathForLocalization(req1, base_path,
        delService);
    Assert.assertFalse(lfs.util().exists(rPath));
    verify(delService, times(1)).delete(eq(user), eq(conflictPath));
  } finally {
    lfs.delete(base_path, true);
    if (dispatcher != null) {
      dispatcher.stop();
    }
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Test
public void testResourcePresentInGoodDir() throws IOException {
  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 cId1 = BuilderUtils.newContainerId(1, 1, 1, 1);
    LocalizerContext lc1 = new LocalizerContext(user, cId1, null);
    LocalResourceRequest req1 =
        createLocalResourceRequest(user, 1, 1, LocalResourceVisibility.PUBLIC);
    LocalResourceRequest req2 =
        createLocalResourceRequest(user, 2, 1, LocalResourceVisibility.PUBLIC);
    LocalizedResource lr1 = createLocalizedResource(req1, dispatcher);
    LocalizedResource lr2 = createLocalizedResource(req2, dispatcher);
    ConcurrentMap<LocalResourceRequest, LocalizedResource> localrsrc =
        new ConcurrentHashMap<LocalResourceRequest, LocalizedResource>();
    localrsrc.put(req1, lr1);
    localrsrc.put(req2, lr2);
    LocalDirsHandlerService dirsHandler = mock(LocalDirsHandlerService.class);
    List<String> goodDirs = new ArrayList<String>();
    // /tmp/somedir2 is bad
    goodDirs.add("/tmp/somedir1/");
    goodDirs.add("/tmp/somedir2");
    Mockito.when(dirsHandler.getLocalDirs()).thenReturn(goodDirs);
    Mockito.when(dirsHandler.getLocalDirsForRead()).thenReturn(goodDirs);
    LocalResourcesTrackerImpl tracker =
        new LocalResourcesTrackerImpl(user, null, dispatcher, localrsrc,
            true , conf, new NMNullStateStoreService(), dirsHandler);
    ResourceEvent req11Event =
        new ResourceRequestEvent(req1, LocalResourceVisibility.PUBLIC, lc1);
    ResourceEvent req21Event =
        new ResourceRequestEvent(req2, LocalResourceVisibility.PUBLIC, lc1);
    // Localize R1 for C1
    tracker.handle(req11Event);
    // Localize R2 for C1
    tracker.handle(req21Event);
    dispatcher.await();
    // Localize resource1
    Path p1 = tracker.getPathForLocalization(req1,
        new Path("/tmp/somedir1"), null);
    Path p2 = tracker.getPathForLocalization(req2,
        new Path("/tmp/somedir2"), null);
    ResourceLocalizedEvent rle1 = new ResourceLocalizedEvent(req1, p1, 1);
    tracker.handle(rle1);
    ResourceLocalizedEvent rle2 = new ResourceLocalizedEvent(req2, p2, 1);
    tracker.handle(rle2);
    dispatcher.await();
    // Remove somedir2 from gooddirs
    Assert.assertTrue(tracker.checkLocalResource(lr2));
    goodDirs.remove(1);
    Assert.assertFalse(tracker.checkLocalResource(lr2));
  } finally {
    if (dispatcher != null) {
      dispatcher.stop();
    }
  }
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
public ContainerEventMatcher(ContainerId idRef, ContainerEventType type) {
  this.idRef = idRef;
  this.type = type;
}
 

import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerEventType; //导入依赖的package包/类
@Test( timeout = 10000)
@SuppressWarnings("unchecked") // mocked generics
public void testLocalizerRunnerException() throws Exception {
  DrainDispatcher dispatcher = new DrainDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  EventHandler<ApplicationEvent> applicationBus = mock(EventHandler.class);
  dispatcher.register(ApplicationEventType.class, applicationBus);
  EventHandler<ContainerEvent> containerBus = mock(EventHandler.class);
  dispatcher.register(ContainerEventType.class, containerBus);

  ContainerExecutor exec = mock(ContainerExecutor.class);
  LocalDirsHandlerService dirsHandler = new LocalDirsHandlerService();
  LocalDirsHandlerService dirsHandlerSpy = spy(dirsHandler);
  dirsHandlerSpy.init(conf);

  DeletionService delServiceReal = new DeletionService(exec);
  DeletionService delService = spy(delServiceReal);
  delService.init(new Configuration());
  delService.start();

  ResourceLocalizationService rawService =
      new ResourceLocalizationService(dispatcher, exec, delService,
      dirsHandlerSpy, nmContext);
  ResourceLocalizationService spyService = spy(rawService);
  doReturn(mockServer).when(spyService).createServer();
  try {
    spyService.init(conf);
    spyService.start();

    // init application
    final Application app = mock(Application.class);
    final ApplicationId appId =
        BuilderUtils.newApplicationId(314159265358979L, 3);
    when(app.getUser()).thenReturn("user0");
    when(app.getAppId()).thenReturn(appId);
    spyService.handle(new ApplicationLocalizationEvent(
        LocalizationEventType.INIT_APPLICATION_RESOURCES, app));
    dispatcher.await();

    Random r = new Random();
    long seed = r.nextLong();
    System.out.println("SEED: " + seed);
    r.setSeed(seed);
    final Container c = getMockContainer(appId, 42, "user0");
    final LocalResource resource1 = getPrivateMockedResource(r);
    System.out.println("Here 4");
    
    final LocalResourceRequest req1 = new LocalResourceRequest(resource1);
    Map<LocalResourceVisibility, Collection<LocalResourceRequest>> rsrcs =
      new HashMap<LocalResourceVisibility, 
                  Collection<LocalResourceRequest>>();
    List<LocalResourceRequest> privateResourceList =
        new ArrayList<LocalResourceRequest>();
    privateResourceList.add(req1);
    rsrcs.put(LocalResourceVisibility.PRIVATE, privateResourceList);

    final Constructor<?>[] constructors =
        FSError.class.getDeclaredConstructors();
    constructors[0].setAccessible(true);
    FSError fsError =
        (FSError) constructors[0].newInstance(new IOException("Disk Error"));

    Mockito
      .doThrow(fsError)
      .when(dirsHandlerSpy)
      .getLocalPathForWrite(isA(String.class));
    spyService.handle(new ContainerLocalizationRequestEvent(c, rsrcs));
    Thread.sleep(1000);
    dispatcher.await();
    // Verify if ContainerResourceFailedEvent is invoked on FSError
    verify(containerBus).handle(isA(ContainerResourceFailedEvent.class));
  } finally {
    spyService.stop();
    dispatcher.stop();
    delService.stop();
  }
}