CombinedSequenceFile extends SequenceFile { @Override public void sourceConfInit( FlowProcess<? extends Configuration> flowProcess, Tap<Configuration, RecordReader, OutputCollector> tap, Configuration conf ) { super.sourceConfInit(flowProcess, tap, conf); updateJobConfForLocalSettings(conf); MapReduceInputFormatWrapper.setWrappedInputFormat(SequenceFileInputFormat.class, conf); DelegateCombineFileInputFormat.setDelegateInputFormat(conf, MapReduceInputFormatWrapper.class); } protected  CombinedSequenceFile();  CombinedSequenceFile(Fields fields); @Override void sourceConfInit(
      FlowProcess<? extends Configuration> flowProcess,
      Tap<Configuration, RecordReader, OutputCollector> tap,
      Configuration conf ); @Override void sinkConfInit( FlowProcess<? extends Configuration> flowProcess, Tap<Configuration, RecordReader, OutputCollector> tap, Configuration conf ); static final String COMPRESS_ENABLE; static final String COMPRESS_TYPE; static final String COMPRESS_CODEC; }
@Test public void testHadoopConf() { CombinedSequenceFile csfScheme = new CombinedSequenceFile(Fields.ALL); JobConf conf = new JobConf(); FlowProcess fp = new HadoopFlowProcess(); Tap<Configuration, RecordReader, OutputCollector> tap = new TempHfs(conf, "test", CombinedSequenceFile.class, false); csfScheme.sourceConfInit(fp, tap, conf); assertEquals( "MapReduceInputFormatWrapper shold wrap mapred.SequenceFileinputFormat", "org.apache.hadoop.mapred.SequenceFileInputFormat", conf.get(MapReduceInputFormatWrapper.CLASS_CONF_KEY) ); assertEquals( "Delegate combiner should wrap MapReduceInputFormatWrapper", "com.twitter.elephantbird.mapreduce.input.MapReduceInputFormatWrapper", conf.get(DelegateCombineFileInputFormat.COMBINED_INPUT_FORMAT_DELEGATE) ); assertEquals( "DeprecatedInputFormatWrapper should wrap Delegate combiner", "com.twitter.elephantbird.mapreduce.input.combine.DelegateCombineFileInputFormat", conf.get(DeprecatedInputFormatWrapper.CLASS_CONF_KEY) ); }