MapReduceEncodingManager extends BaseEncodingManager { public void raidFiles(PolicyInfo info) throws IOException { MapReduceEncoder dr = new MapReduceEncoder(conf); boolean started = dr.startDistRaid(info); if (started) { String path = info.getSrcPath().toUri().getPath(); persistActiveJob(path, dr.getJobID(), dr.getConf().get(MapReduceEncoder.JOB_DIR_LABEL)); currentJobs.put(path, dr); } } MapReduceEncodingManager(Configuration conf); @Override void encodeFile(EncodingPolicy policy, Path sourceFile,
      Path parityFile, boolean copy); void raidFiles(PolicyInfo info); @Override List<Report> computeReports(); @Override void cancelAll(); @Override void cancel(String toCancel); static final Log LOG; static final String ENCODING_JOB_EXECUTION_LIMIT; static final long DEFAULT_ENCODING_JOB_EXECUTION_LIMIT; }
@Test public void testRaidFiles() throws IOException, InterruptedException { DistributedFileSystem dfs = (DistributedFileSystem) getFileSystem(); MapReduceEncodingManager encodingManager = new MapReduceEncodingManager(conf); Util.createRandomFile(dfs, testFile, seed, TEST_BLOCK_COUNT, DFS_TEST_BLOCK_SIZE); Codec.initializeCodecs(conf); EncodingPolicy policy = new EncodingPolicy("src", (short) 1); encodingManager.encodeFile(policy, testFile, parityFile, false); List<Report> reports; while ((reports = encodingManager.computeReports()).size() > 0) { Assert.assertNotSame("Encoding Assert.failed.", Report.Status.FAILED, reports.get(0).getStatus()); Thread.sleep(1000); } FileStatus parityStatus = dfs.getFileStatus(parityFile); Assert.assertEquals(parityStatus.getLen(), 6 * DFS_TEST_BLOCK_SIZE); try { FSDataInputStream in = dfs.open(parityFile); byte[] buff = new byte[6 * DFS_TEST_BLOCK_SIZE]; in.readFully(0, buff); } catch (BlockMissingException e) { LOG.error("Reading parity Assert.failed", e); Assert.fail("Parity could not be read."); } }