AbstractYarnScheduler extends AbstractService implements ResourceScheduler { @Override public RMContainer getRMContainer(ContainerId containerId) { SchedulerApplicationAttempt attempt = getCurrentAttemptForContainer(containerId); return (attempt == null) ? null : attempt.getRMContainer(containerId); } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); @VisibleForTesting ClusterNodeTracker getNodeTracker(); @VisibleForTesting SchedulingMonitorManager getSchedulingMonitorManager(); List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); List<N> getBlacklistedNodes(final SchedulerApplicationAttempt app); List<N> getNodes(final NodeFilter filter); boolean shouldContainersBeAutoUpdated(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); @Override Resource getMaximumResourceCapability(String queueName); SchedulerHealth getSchedulerHealth(); long getLastNodeUpdateTime(); T getApplicationAttempt(ApplicationAttemptId applicationAttemptId); @Override SchedulerAppReport getSchedulerAppInfo(
      ApplicationAttemptId appAttemptId); @Override ApplicationResourceUsageReport getAppResourceUsageReport(
      ApplicationAttemptId appAttemptId); T getCurrentAttemptForContainer(ContainerId containerId); @Override RMContainer getRMContainer(ContainerId containerId); @Override SchedulerNodeReport getNodeReport(NodeId nodeId); @Override String moveApplication(ApplicationId appId, String newQueue); @Override void preValidateMoveApplication(ApplicationId appId,
      String newQueue); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); void recoverContainersOnNode(List<NMContainerStatus> containerReports,
      RMNode nm); @VisibleForTesting void clearPendingContainerCache(); @VisibleForTesting @Private // clean up a completed container void completedContainer(RMContainer rmContainer,
      ContainerStatus containerStatus, RMContainerEventType event); @Override N getSchedulerNode(NodeId nodeId); @Override void moveAllApps(String sourceQueue, String destQueue); @Override void killAllAppsInQueue(String queueName); void updateNodeResource(RMNode nm,
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); @Override List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId); @Override List<SchedulingRequest> getPendingSchedulingRequestsForAttempt(
      ApplicationAttemptId attemptId); @Override Priority checkAndGetApplicationPriority(
      Priority priorityRequestedByApp, UserGroupInformation user,
      String queueName, ApplicationId applicationId); @Override Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user); @Override Priority getMaxClusterLevelAppPriority(); @Override void setClusterMaxPriority(Configuration conf); ActivitiesManager getActivitiesManager(); Clock getClock(); @VisibleForTesting void setClock(Clock clock); @Lock(Lock.NoLock.class) SchedulerNode getNode(NodeId nodeId); @Override Resource getNormalizedResource(Resource requestedResource,
                                        Resource maxResourceCapability); @Override List<NodeId> getNodeIds(String resourceName); void asyncContainerRelease(RMContainer container); Resource getMinimumAllocation(); Resource getMaximumAllocation(); @Override long checkAndGetApplicationLifetime(String queueName, long lifetime); @Override long getMaximumApplicationLifetime(String queueName); @VisibleForTesting abstract void killContainer(RMContainer container); @VisibleForTesting void update(); @Override void reinitialize(Configuration conf, RMContext rmContext); @Override boolean attemptAllocationOnNode(SchedulerApplicationAttempt appAttempt,
      SchedulingRequest schedulingRequest, SchedulerNode schedulerNode); @Override void resetSchedulerMetrics(); @Lock(Lock.NoLock.class) List<N> getAllNodes();  }
@Test(timeout=60000) public void testContainerRecoveredByNode() throws Exception { System.out.println("Starting testContainerRecoveredByNode"); final int maxMemory = 10 * 1024; YarnConfiguration conf = getConf(); conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true); conf.setBoolean( YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED, true); conf.set( YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName()); MockRM rm1 = new MockRM(conf); try { rm1.start(); RMApp app1 = rm1.submitApp(200, "name", "user", new HashMap<ApplicationAccessType, String>(), false, "default", -1, null, "Test", false, true); MockNM nm1 = new MockNM("127.0.0.1:1234", 10240, rm1.getResourceTrackerService()); nm1.registerNode(); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm1, nm1); am1.allocate("127.0.0.1", 8192, 1, new ArrayList<ContainerId>()); YarnScheduler scheduler = rm1.getResourceScheduler(); RMNode node1 = MockNodes.newNodeInfo( 0, Resources.createResource(maxMemory), 1, "127.0.0.2"); ContainerId containerId = ContainerId.newContainerId( app1.getCurrentAppAttempt().getAppAttemptId(), 2); NMContainerStatus containerReport = NMContainerStatus.newInstance(containerId, 0, ContainerState.RUNNING, Resource.newInstance(1024, 1), "recover container", 0, Priority.newInstance(0), 0); List<NMContainerStatus> containerReports = new ArrayList<>(); containerReports.add(containerReport); scheduler.handle(new NodeAddedSchedulerEvent(node1, containerReports)); RMContainer rmContainer = scheduler.getRMContainer(containerId); Assert.assertEquals(app1.getQueue(), rmContainer.getQueueName()); } finally { rm1.stop(); System.out.println("Stopping testContainerRecoveredByNode"); } }