@SuppressWarnings("deprecation")
  public void runLocalization(final InetSocketAddress nmAddr)
      throws IOException, InterruptedException {
    // load credentials
    initDirs(conf, user, appId, lfs, localDirs);
    final Credentials creds = new Credentials();
    DataInputStream credFile = null;
    try {
      // assume credentials in cwd
      // TODO: Fix
      Path tokenPath = new Path(tokenFileName);
      credFile = lfs.open(tokenPath);
      creds.readTokenStorageStream(credFile);
      // Explicitly deleting token file.
      lfs.delete(tokenPath, false);      
    } finally  {
      if (credFile != null) {
        credFile.close();
      }
    }
    // create localizer context
    UserGroupInformation remoteUser =
      UserGroupInformation.createRemoteUser(user);
    remoteUser.addToken(creds.getToken(LocalizerTokenIdentifier.KIND));
    final LocalizationProtocol nodeManager =
        remoteUser.doAs(new PrivilegedAction<LocalizationProtocol>() {
          @Override
          public LocalizationProtocol run() {
            return getProxy(nmAddr);
          }
        });

    // create user context
    UserGroupInformation ugi =
      UserGroupInformation.createRemoteUser(user);
    for (Token<? extends TokenIdentifier> token : creds.getAllTokens()) {
      ugi.addToken(token);
    }

    ExecutorService exec = null;
    try {
      exec = createDownloadThreadPool();
      CompletionService<Path> ecs = createCompletionService(exec);
      localizeFiles(nodeManager, ecs, ugi);
    } catch (Throwable e) {
      throw new IOException(e);
    } finally {
      try {
        if (exec != null) {
          exec.shutdown();
          destroyShellProcesses(getAllShells());
          exec.awaitTermination(10, TimeUnit.SECONDS);
        }
        LocalDirAllocator.removeContext(appCacheDirContextName);
      } finally {
        closeFileSystems(ugi);
      }
    }
  }
@Test(timeout = 15000)
  public void testMainFailure() throws Exception {
    ContainerLocalizerWrapper wrapper = new ContainerLocalizerWrapper();
    ContainerLocalizer localizer = wrapper.setupContainerLocalizerForTest();
    LocalizationProtocol nmProxy = wrapper.nmProxy;
    mockOutDownloads(localizer);

    // Assume the NM heartbeat fails say because of absent tokens.
    when(nmProxy.heartbeat(isA(LocalizerStatus.class))).thenThrow(
        new YarnException("Sigh, no token!"));

    // run localization, it should fail
    try {
      localizer.runLocalization(nmAddr);
      Assert.fail("Localization succeeded unexpectedly!");
    } catch (IOException e) {
      Assert.assertTrue(e.getMessage().contains("Sigh, no token!"));
    }
  }