@VisibleForTesting
  protected Path maybeDownloadJars(String sName, String className, String
      remoteFile, AuxServiceFile.TypeEnum type, Configuration conf)
      throws IOException {
    // load AuxiliaryService from remote classpath
    FileContext localLFS = getLocalFileContext(conf);
    // create NM aux-service dir in NM localdir if it does not exist.
    Path nmAuxDir = dirsHandler.getLocalPathForWrite("."
        + Path.SEPARATOR + NM_AUX_SERVICE_DIR);
    if (!localLFS.util().exists(nmAuxDir)) {
      try {
        localLFS.mkdir(nmAuxDir, NM_AUX_SERVICE_DIR_PERM, true);
      } catch (IOException ex) {
        throw new YarnRuntimeException("Fail to create dir:"
            + nmAuxDir.toString(), ex);
      }
    }
    Path src = new Path(remoteFile);
    FileContext remoteLFS = getRemoteFileContext(src.toUri(), conf);
    FileStatus scFileStatus = remoteLFS.getFileStatus(src);
    if (!scFileStatus.getOwner().equals(
        this.userUGI.getShortUserName())) {
      throw new YarnRuntimeException("The remote jarfile owner:"
          + scFileStatus.getOwner() + " is not the same as the NM user:"
          + this.userUGI.getShortUserName() + ".");
    }
    if ((scFileStatus.getPermission().toShort() & 0022) != 0) {
      throw new YarnRuntimeException("The remote jarfile should not "
          + "be writable by group or others. "
          + "The current Permission is "
          + scFileStatus.getPermission().toShort());
    }
    Path downloadDest = new Path(nmAuxDir,
        className + "_" + scFileStatus.getModificationTime());
    // check whether we need to re-download the jar
    // from remote directory
    Path targetDirPath = new Path(downloadDest,
        scFileStatus.getPath().getName());
    FileStatus[] allSubDirs = localLFS.util().listStatus(nmAuxDir);
    for (FileStatus sub : allSubDirs) {
      if (sub.getPath().getName().equals(downloadDest.getName())) {
        return targetDirPath;
      } else {
        if (sub.getPath().getName().contains(className) &&
            !sub.getPath().getName().endsWith(DEL_SUFFIX)) {
          Path delPath = new Path(sub.getPath().getParent(),
              sub.getPath().getName() + DEL_SUFFIX);
          localLFS.rename(sub.getPath(), delPath);
          LOG.info("delete old aux service jar dir:"
              + delPath.toString());
          FileDeletionTask deletionTask = new FileDeletionTask(
              this.delService, null, delPath, null);
          this.delService.delete(deletionTask);
        }
      }
    }
    LocalResourceType srcType;
    if (type == AuxServiceFile.TypeEnum.STATIC) {
      srcType = LocalResourceType.FILE;
    } else if (type == AuxServiceFile.TypeEnum.ARCHIVE) {
      srcType = LocalResourceType.ARCHIVE;
    } else {
      throw new YarnRuntimeException(
          "Cannot unpack file of type " + type + " from remote-file-path:" +
              src + "for aux-service:" + ".\n");
    }
    LocalResource scRsrc = LocalResource.newInstance(
        URL.fromURI(src.toUri()),
        srcType, LocalResourceVisibility.PRIVATE,
        scFileStatus.getLen(), scFileStatus.getModificationTime());
    FSDownload download = new FSDownload(localLFS, null, conf,
        downloadDest, scRsrc, null);
    try {
      // don't need to convert downloaded path into a dir
      // since it's already a jar path.
      return download.call();
    } catch (Exception ex) {
      throw new YarnRuntimeException(
          "Exception happend while downloading files "
              + "for aux-service:" + sName + " and remote-file-path:"
              + src + ".\n" + ex.getMessage());
    }
  }
@Test (timeout = 15000)
  public void testReuseLocalizedAuxiliaryJar() throws Exception {
    File testJar = null;
    AuxServices aux = null;
    Configuration conf = new YarnConfiguration();
    FileSystem fs = FileSystem.get(conf);
    String root = "target/LocalDir";
    try {
      testJar = JarFinder.makeClassLoaderTestJar(this.getClass(), rootDir,
          "test-runjar.jar", 2048, ServiceB.class.getName(), LightService
          .class.getName());
      Context mockContext = mock(Context.class);
      LocalDirsHandlerService mockDirsHandler = mock(
          LocalDirsHandlerService.class);
      Path rootAuxServiceDirPath = new Path(root, "nmAuxService");
      when(mockDirsHandler.getLocalPathForWrite(anyString())).thenReturn(
          rootAuxServiceDirPath);
      when(mockContext.getLocalDirsHandler()).thenReturn(mockDirsHandler);
      aux = new AuxServices(MOCK_AUX_PATH_HANDLER, mockContext,
          MOCK_DEL_SERVICE);
      // First Time the jar gets localized
      Path path = aux.maybeDownloadJars("ServiceB", ServiceB.class.getName(),
          testJar.getAbsolutePath(), AuxServiceFile.TypeEnum.STATIC, conf);

      // Validate the path on reuse of localized jar
      path = aux.maybeDownloadJars("ServiceB", ServiceB.class.getName(),
          testJar.getAbsolutePath(), AuxServiceFile.TypeEnum.STATIC, conf);
      assertFalse("Failed to reuse the localized jar",
          path.toString().endsWith("/*"));
    } finally {
      if (testJar != null) {
        testJar.delete();
      }
      if (fs.exists(new Path(root))) {
        fs.delete(new Path(root), true);
      }
    }
  }