@SuppressWarnings("unchecked") @Test public void testInitNextRecordReader() throws IOException{ JobConf conf = new JobConf(); Path[] paths = new Path[3]; long[] fileLength = new long[3]; File[] files = new File[3]; LongWritable key = new LongWritable(1); Text value = new Text(); try { for(int i=0;i<3;i++){ fileLength[i] = i; File dir = new File(outDir.toString()); dir.mkdir(); files[i] = new File(dir,"testfile"+i); FileWriter fileWriter = new FileWriter(files[i]); fileWriter.close(); paths[i] = new Path(outDir+"/testfile"+i); } CombineFileSplit combineFileSplit = new CombineFileSplit(conf, paths, fileLength); Reporter reporter = Mockito.mock(Reporter.class); CombineFileRecordReader cfrr = new CombineFileRecordReader(conf, combineFileSplit, reporter, TextRecordReaderWrapper.class); verify(reporter).progress(); Assert.assertFalse(cfrr.next(key,value)); verify(reporter, times(3)).progress(); } finally { FileUtil.fullyDelete(new File(outDir.toString())); } }
protected boolean initNextRecordReader() throws IOException { if (curReader != null) { curReader.close(); curReader = null; if (idx > 0) { progress += split.getLength(idx-1); } } if (idx == split.getNumPaths()) { return false; } reporter.progress(); try { curReader = rrConstructor.newInstance(new Object [] {split, jc, reporter, Integer.valueOf(idx)}); jc.set(JobContext.MAP_INPUT_FILE, split.getPath(idx).toString()); jc.setLong(JobContext.MAP_INPUT_START, split.getOffset(idx)); jc.setLong(JobContext.MAP_INPUT_PATH, split.getLength(idx)); } catch (Exception e) { throw new RuntimeException (e); } idx++; return true; }
CombineFileRecordReader implements RecordReader<K, V> { protected boolean initNextRecordReader() throws IOException { if (curReader != null) { curReader.close(); curReader = null; if (idx > 0) { progress += split.getLength(idx-1); } } if (idx == split.getNumPaths()) { return false; } reporter.progress(); try { curReader = rrConstructor.newInstance(new Object [] {split, jc, reporter, Integer.valueOf(idx)}); jc.set(JobContext.MAP_INPUT_FILE, split.getPath(idx).toString()); jc.setLong(JobContext.MAP_INPUT_START, split.getOffset(idx)); jc.setLong(JobContext.MAP_INPUT_PATH, split.getLength(idx)); } catch (Exception e) { throw new RuntimeException (e); } idx++; return true; } }
CombineFileRecordReader implements RecordReader<K, V> { protected boolean initNextRecordReader() throws IOException { if (curReader != null) { curReader.close(); curReader = null; if (idx > 0) { progress += split.getLength(idx-1); } } if (idx == split.getNumPaths()) { return false; } reporter.progress(); try { curReader = rrConstructor.newInstance(new Object [] {split, jc, reporter, Integer.valueOf(idx)}); jc.set(JobContext.MAP_INPUT_FILE, split.getPath(idx).toString()); jc.setLong(JobContext.MAP_INPUT_START, split.getOffset(idx)); jc.setLong(JobContext.MAP_INPUT_PATH, split.getLength(idx)); } catch (Exception e) { throw new RuntimeException (e); } idx++; return true; } CombineFileRecordReader(JobConf job, CombineFileSplit split, 
                                 Reporter reporter,
                                 Class<RecordReader<K, V>> rrClass); }
CombineFileRecordReader implements RecordReader<K, V> { protected boolean initNextRecordReader() throws IOException { if (curReader != null) { curReader.close(); curReader = null; if (idx > 0) { progress += split.getLength(idx-1); } } if (idx == split.getNumPaths()) { return false; } reporter.progress(); try { curReader = rrConstructor.newInstance(new Object [] {split, jc, reporter, Integer.valueOf(idx)}); jc.set(JobContext.MAP_INPUT_FILE, split.getPath(idx).toString()); jc.setLong(JobContext.MAP_INPUT_START, split.getOffset(idx)); jc.setLong(JobContext.MAP_INPUT_PATH, split.getLength(idx)); } catch (Exception e) { throw new RuntimeException (e); } idx++; return true; } CombineFileRecordReader(JobConf job, CombineFileSplit split, 
                                 Reporter reporter,
                                 Class<RecordReader<K, V>> rrClass); boolean next(K key, V value); K createKey(); V createValue(); long getPos(); void close(); float getProgress(); LineReader getLineReader(); }
CombineFileRecordReader implements RecordReader<K, V> { protected boolean initNextRecordReader() throws IOException { if (curReader != null) { curReader.close(); curReader = null; if (idx > 0) { progress += split.getLength(idx-1); } } if (idx == split.getNumPaths()) { return false; } reporter.progress(); try { curReader = rrConstructor.newInstance(new Object [] {split, jc, reporter, Integer.valueOf(idx)}); jc.set(JobContext.MAP_INPUT_FILE, split.getPath(idx).toString()); jc.setLong(JobContext.MAP_INPUT_START, split.getOffset(idx)); jc.setLong(JobContext.MAP_INPUT_PATH, split.getLength(idx)); } catch (Exception e) { throw new RuntimeException (e); } idx++; return true; } CombineFileRecordReader(JobConf job, CombineFileSplit split, 
                                 Reporter reporter,
                                 Class<RecordReader<K, V>> rrClass); boolean next(K key, V value); K createKey(); V createValue(); long getPos(); void close(); float getProgress(); LineReader getLineReader();  }