@Override
  public RMContainer getRMContainer(ContainerId containerId) {
    SchedulerApplicationAttempt attempt =
        getCurrentAttemptForContainer(containerId);
    return (attempt == null) ? null : attempt.getRMContainer(containerId);
  }
@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();
      MockRMAppSubmissionData data =
          MockRMAppSubmissionData.Builder.createWithMemory(200, rm1)
          .withAppName("name")
          .withUser("user")
          .withAcls(new HashMap<ApplicationAccessType, String>())
          .withUnmanagedAM(false)
          .withQueue("default")
          .withMaxAppAttempts(-1)
          .withCredentials(null)
          .withAppType("Test")
          .withWaitForAppAcceptedState(false)
          .withKeepContainers(true)
          .build();
      RMApp app1 =
          MockRMAppSubmitter.submit(rm1, data);
      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);

      //verify queue name when rmContainer is recovered
      if (scheduler instanceof CapacityScheduler) {
        Assert.assertEquals(
            app1.getQueue(),
            rmContainer.getQueueName());
      } else {
        Assert.assertEquals(app1.getQueue(), rmContainer.getQueueName());
      }

    } finally {
      rm1.stop();
      System.out.println("Stopping testContainerRecoveredByNode");
    }
  }