public static void determineTimestamps(Configuration job,
      Map<URI, FileStatus> statCache) throws IOException {
    URI[] tarchives = JobContextImpl.getCacheArchives(job);
    if (tarchives != null) {
      FileStatus status = getFileStatus(job, tarchives[0], statCache);
      StringBuilder archiveFileSizes =
        new StringBuilder(String.valueOf(status.getLen()));
      StringBuilder archiveTimestamps =
        new StringBuilder(String.valueOf(status.getModificationTime()));
      for (int i = 1; i < tarchives.length; i++) {
        status = getFileStatus(job, tarchives[i], statCache);
        archiveFileSizes.append(",");
        archiveFileSizes.append(String.valueOf(status.getLen()));
        archiveTimestamps.append(",");
        archiveTimestamps.append(String.valueOf(status.getModificationTime()));
      }
      job.set(MRJobConfig.CACHE_ARCHIVES_SIZES, archiveFileSizes.toString());
      setArchiveTimestamps(job, archiveTimestamps.toString());
    }
  
    URI[] tfiles = JobContextImpl.getCacheFiles(job);
    if (tfiles != null) {
      FileStatus status = getFileStatus(job, tfiles[0], statCache);
      StringBuilder fileSizes =
        new StringBuilder(String.valueOf(status.getLen()));
      StringBuilder fileTimestamps = new StringBuilder(String.valueOf(
        status.getModificationTime()));
      for (int i = 1; i < tfiles.length; i++) {
        status = getFileStatus(job, tfiles[i], statCache);
        fileSizes.append(",");
        fileSizes.append(String.valueOf(status.getLen()));
        fileTimestamps.append(",");
        fileTimestamps.append(String.valueOf(status.getModificationTime()));
      }
      job.set(MRJobConfig.CACHE_FILES_SIZES, fileSizes.toString());
      setFileTimestamps(job, fileTimestamps.toString());
    }
  }
@Test
  public void testDetermineTimestamps() throws IOException {
    Job job = Job.getInstance(conf);
    job.addCacheFile(firstCacheFile.toUri());
    job.addCacheFile(secondCacheFile.toUri());
    Configuration jobConf = job.getConfiguration();
    
    Map<URI, FileStatus> statCache = new HashMap<>();
    ClientDistributedCacheManager.determineTimestamps(jobConf, statCache);
    
    FileStatus firstStatus = statCache.get(firstCacheFile.toUri());
    FileStatus secondStatus = statCache.get(secondCacheFile.toUri());
    
    assertNotNull(firstCacheFile + " was not found in the stats cache",
        firstStatus);
    assertNotNull(secondCacheFile + " was not found in the stats cache",
        secondStatus);
    assertEquals("Missing/extra entries found in the stats cache",
        2, statCache.size());
    String expected = firstStatus.getModificationTime() + ","
        + secondStatus.getModificationTime();
    assertEquals(expected, jobConf.get(MRJobConfig.CACHE_FILE_TIMESTAMPS));

    job = Job.getInstance(conf);
    job.addCacheFile(new Path(TEST_VISIBILITY_CHILD_DIR, "*").toUri());
    jobConf = job.getConfiguration();
    statCache.clear();
    ClientDistributedCacheManager.determineTimestamps(jobConf, statCache);

    FileStatus thirdStatus = statCache.get(TEST_VISIBILITY_CHILD_DIR.toUri());

    assertEquals("Missing/extra entries found in the stats cache",
        1, statCache.size());
    assertNotNull(TEST_VISIBILITY_CHILD_DIR
        + " was not found in the stats cache", thirdStatus);
    expected = Long.toString(thirdStatus.getModificationTime());
    assertEquals("Incorrect timestamp for " + TEST_VISIBILITY_CHILD_DIR,
        expected, jobConf.get(MRJobConfig.CACHE_FILE_TIMESTAMPS));
  }