@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
  public void testMultipleLocalizers() throws Exception {
    FakeContainerLocalizerWrapper testA = new FakeContainerLocalizerWrapper();
    FakeContainerLocalizerWrapper testB = new FakeContainerLocalizerWrapper();

    FakeContainerLocalizer localizerA = testA.init();
    FakeContainerLocalizer localizerB = testB.init();

    // run localization
    Thread threadA = new Thread() {
      @Override
      public void run() {
        try {
          localizerA.runLocalization(nmAddr);
        } catch (Exception e) {
          LOG.warn(e.toString());
        }
      }
    };
    Thread threadB = new Thread() {
      @Override
      public void run() {
        try {
          localizerB.runLocalization(nmAddr);
        } catch (Exception e) {
          LOG.warn(e.toString());
        }
      }
    };
    ShellCommandExecutor shexcA = null;
    ShellCommandExecutor shexcB = null;
    try {
      threadA.start();
      threadB.start();

      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          FakeContainerLocalizer.FakeLongDownload downloader =
              localizerA.getDownloader();
          return downloader != null && downloader.getShexc() != null &&
              downloader.getShexc().getProcess() != null;
        }
      }, 10, 30000);

      GenericTestUtils.waitFor(new Supplier<Boolean>() {
        @Override
        public Boolean get() {
          FakeContainerLocalizer.FakeLongDownload downloader =
              localizerB.getDownloader();
          return downloader != null && downloader.getShexc() != null &&
              downloader.getShexc().getProcess() != null;
        }
      }, 10, 30000);

      shexcA = localizerA.getDownloader().getShexc();
      shexcB = localizerB.getDownloader().getShexc();

      assertTrue("Localizer A process not running, but should be",
          shexcA.getProcess().isAlive());
      assertTrue("Localizer B process not running, but should be",
          shexcB.getProcess().isAlive());

      // Stop heartbeat from giving anymore resources to download
      testA.heartbeatResponse++;
      testB.heartbeatResponse++;

      // Send DIE to localizerA. This should kill its subprocesses
      testA.heartbeatResponse++;

      threadA.join();
      shexcA.getProcess().waitFor(10000, TimeUnit.MILLISECONDS);

      assertFalse("Localizer A process is still running, but shouldn't be",
          shexcA.getProcess().isAlive());
      assertTrue("Localizer B process not running, but should be",
          shexcB.getProcess().isAlive());

    } finally {
      // Make sure everything gets cleaned up
      // Process A should already be dead
      shexcA.getProcess().destroy();
      shexcB.getProcess().destroy();
      shexcA.getProcess().waitFor(10000, TimeUnit.MILLISECONDS);
      shexcB.getProcess().waitFor(10000, TimeUnit.MILLISECONDS);

      threadA.join();
      // Send DIE to localizer B
      testB.heartbeatResponse++;
      threadB.join();
    }
  }