@VisibleForTesting
  void updateLocalizationStatuses(
      List<org.apache.hadoop.yarn.api.records.LocalizationStatus> statuses) {
    Map<String, String> resourcesCpy = new HashMap<>();
    readLock.lock();
    try {
      if (resolvedParams == null || resolvedParams.didLaunchFail() ||
          resolvedParams.getResolvedRsrcPaths() == null ||
          resolvedParams.getResolvedRsrcPaths().isEmpty()) {
        cancelLclRetriever();
        return;
      }
      resourcesCpy.putAll(resolvedParams.getResolvedRsrcPaths());
    } finally {
      readLock.unlock();
    }
    boolean allCompleted = true;
    Map<String, LocalizationStatus> fromNM = new HashMap<>();
    statuses.forEach(statusFromNM -> {
      LocalizationStatus lstatus = new LocalizationStatus()
          .destFile(statusFromNM.getResourceKey())
          .diagnostics(statusFromNM.getDiagnostics())
          .state(statusFromNM.getLocalizationState());
      fromNM.put(statusFromNM.getResourceKey(), lstatus);
    });

    for (String resourceKey : resourcesCpy.keySet()) {
      LocalizationStatus lstatus = fromNM.get(resourceKey);
      if (lstatus == null ||
          lstatus.getState().equals(LocalizationState.PENDING)) {
        allCompleted = false;
        break;
      }
    }

    List<LocalizationStatus> statusList = new ArrayList<>();
    statusList.addAll(fromNM.values());
    this.containerSpec.setLocalizationStatuses(statusList);
    if (allCompleted) {
      cancelLclRetriever();
    }
  }
@Test
  public void testUpdateLocalizationStatuses() throws Exception {
    Service def = TestServiceManager.createBaseDef(
        "testUpdateLocalizationStatuses");

    String file1 = rule.getServiceBasePath().toString() + "/file1";
    Files.write(Paths.get(file1), "test file".getBytes(),
        StandardOpenOption.CREATE_NEW);

    org.apache.hadoop.yarn.service.api.records.Component compDef =
        def.getComponents().iterator().next();
    ConfigFile configFile1 = new ConfigFile();
    configFile1.setType(ConfigFile.TypeEnum.STATIC);
    configFile1.setSrcFile(file1);
    compDef.setConfiguration(new Configuration().files(
        Lists.newArrayList(configFile1)));

    ServiceContext context = new MockRunningServiceContext(rule, def);
    Component component = context.scheduler.getAllComponents().get(
        compDef.getName());
    ComponentInstance instance = component.getAllComponentInstances().iterator()
        .next();
    LocalizationStatus status = LocalizationStatus.newInstance("file1",
        LocalizationState.PENDING);

    instance.updateLocalizationStatuses(Lists.newArrayList(status));
    Assert.assertTrue("retriever should still be active",
        instance.isLclRetrieverActive());

    Container container = instance.getContainerSpec();
    Assert.assertTrue(container.getLocalizationStatuses() != null);
    Assert.assertEquals("dest file",
        container.getLocalizationStatuses().get(0).getDestFile(),
        status.getResourceKey());
    Assert.assertEquals("state",
        container.getLocalizationStatuses().get(0).getState(),
        status.getLocalizationState());

    status = LocalizationStatus.newInstance("file1",
        LocalizationState.COMPLETED);
    instance.updateLocalizationStatuses(Lists.newArrayList(status));
    Assert.assertTrue("retriever should not be active",
        !instance.isLclRetrieverActive());
    Assert.assertTrue(container.getLocalizationStatuses() != null);
    Assert.assertEquals("dest file",
        container.getLocalizationStatuses().get(0).getDestFile(),
        status.getResourceKey());
    Assert.assertEquals("state",
        container.getLocalizationStatuses().get(0).getState(),
        status.getLocalizationState());
  }