FilePool { public void refresh() throws IOException { updateLock.writeLock().lock(); try { root = new InnerDesc(fs, fs.getFileStatus(path), new MinFileFilter(conf.getLong(GRIDMIX_MIN_FILE, 128 * 1024 * 1024), conf.getLong(GRIDMIX_MAX_TOTAL, 100L * (1L << 40)))); if (0 == root.getSize()) { throw new IOException("Found no satisfactory file in " + path); } } finally { updateLock.writeLock().unlock(); } } FilePool(Configuration conf, Path input); long getInputFiles(long minSize, Collection<FileStatus> files); void refresh(); BlockLocation[] locationsFor(FileStatus stat, long start, long len); static final Logger LOG; static final String GRIDMIX_MIN_FILE; static final String GRIDMIX_MAX_TOTAL; }
@Test public void testUnsuitable() throws Exception { try { final Configuration conf = new Configuration(); conf.setLong(FilePool.GRIDMIX_MIN_FILE, 14 * 1024); final FilePool pool = new FilePool(conf, base); pool.refresh(); } catch (IOException e) { return; } fail(); }
@Test public void testStriper() throws Exception { final Random r = new Random(); final Configuration conf = new Configuration(); final FileSystem fs = FileSystem.getLocal(conf).getRaw(); conf.setLong(FilePool.GRIDMIX_MIN_FILE, 3 * 1024); final FilePool pool = new FilePool(conf, base) { @Override public BlockLocation[] locationsFor(FileStatus stat, long start, long len) throws IOException { return new BlockLocation[] { new BlockLocation() }; } }; pool.refresh(); final int expectedPoolSize = (NFILES / 2 * (NFILES / 2 + 1) - 6) * 1024; final InputStriper striper = new InputStriper(pool, expectedPoolSize); int last = 0; for (int i = 0; i < expectedPoolSize; last = Math.min(expectedPoolSize - i, r.nextInt(expectedPoolSize))) { checkSplitEq(fs, striper.splitFor(pool, last, 0), last); i += last; } final InputStriper striper2 = new InputStriper(pool, expectedPoolSize); checkSplitEq(fs, striper2.splitFor(pool, expectedPoolSize, 0), expectedPoolSize); }