FileInputFormat extends InputFormat<K, V> { public static FileStatus shrinkStatus(FileStatus origStat) { if (origStat.isDirectory() || origStat.getLen() == 0 || !(origStat instanceof LocatedFileStatus)) { return origStat; } else { BlockLocation[] blockLocations = ((LocatedFileStatus)origStat).getBlockLocations(); BlockLocation[] locs = new BlockLocation[blockLocations.length]; int i = 0; for (BlockLocation location : blockLocations) { locs[i++] = new BlockLocation(location); } LocatedFileStatus newStat = new LocatedFileStatus(origStat, locs); return newStat; } }  static void setInputDirRecursive(Job job,
      boolean inputDirRecursive); static boolean getInputDirRecursive(JobContext job); static void setInputPathFilter(Job job,
                                        Class<? extends PathFilter> filter); static void setMinInputSplitSize(Job job,
                                          long size); static long getMinSplitSize(JobContext job); static void setMaxInputSplitSize(Job job,
                                          long size); static long getMaxSplitSize(JobContext context); static PathFilter getInputPathFilter(JobContext context); static FileStatus shrinkStatus(FileStatus origStat); List<InputSplit> getSplits(JobContext job); static void setInputPaths(Job job, 
                                   String commaSeparatedPaths
                                   ); static void addInputPaths(Job job, 
                                   String commaSeparatedPaths
                                   ); static void setInputPaths(Job job, 
                                   Path... inputPaths); static void addInputPath(Job job, 
                                  Path path); static Path[] getInputPaths(JobContext context); static final String INPUT_DIR; static final String SPLIT_MAXSIZE; static final String SPLIT_MINSIZE; static final String PATHFILTER_CLASS; static final String NUM_INPUT_FILES; static final String INPUT_DIR_RECURSIVE; static final String INPUT_DIR_NONRECURSIVE_IGNORE_SUBDIRS; static final String LIST_STATUS_NUM_THREADS; static final int DEFAULT_LIST_STATUS_NUM_THREADS; }
@Test public void testShrinkStatus() throws IOException { Configuration conf = getConfiguration(); MockFileSystem mockFs = (MockFileSystem) new Path("test: Path dir1 = new Path("test:/a1"); RemoteIterator<LocatedFileStatus> statuses = mockFs.listLocatedStatus(dir1); boolean verified = false; while (statuses.hasNext()) { LocatedFileStatus orig = statuses.next(); LocatedFileStatus shrink = (LocatedFileStatus)FileInputFormat.shrinkStatus(orig); Assert.assertTrue(orig.equals(shrink)); if (shrink.getBlockLocations() != null) { Assert.assertEquals(orig.getBlockLocations().length, shrink.getBlockLocations().length); for (int i = 0; i < shrink.getBlockLocations().length; i++) { verified = true; BlockLocation location = shrink.getBlockLocations()[i]; BlockLocation actual = orig.getBlockLocations()[i]; Assert.assertNotNull(((HdfsBlockLocation)actual).getLocatedBlock()); Assert.assertEquals(BlockLocation.class.getName(), location.getClass().getName()); Assert.assertArrayEquals(actual.getHosts(), location.getHosts()); Assert.assertArrayEquals(actual.getCachedHosts(), location.getCachedHosts()); Assert.assertArrayEquals(actual.getStorageIds(), location.getStorageIds()); Assert.assertArrayEquals(actual.getStorageTypes(), location.getStorageTypes()); Assert.assertArrayEquals(actual.getTopologyPaths(), location.getTopologyPaths()); Assert.assertArrayEquals(actual.getNames(), location.getNames()); Assert.assertEquals(actual.getLength(), location.getLength()); Assert.assertEquals(actual.getOffset(), location.getOffset()); Assert.assertEquals(actual.isCorrupt(), location.isCorrupt()); } } else { Assert.assertTrue(orig.getBlockLocations() == null); } } Assert.assertTrue(verified); }