@@ -0,0 +1,1238 @@
+package com.twitter.elephantbird.mapreduce.input;%0A%0Aimport java.io.IOException;%0A%0Aimport org.apache.hadoop.mapreduce.InputSplit;%0Aimport org.apache.hadoop.mapreduce.RecordReader;%0Aimport org.apache.hadoop.mapreduce.TaskAttemptContext;%0A%0A/**%0A * A RecordReader equivalent of FilterInputStream%0A */%0Apublic class FilterRecordReader%3CK, V%3E extends RecordReader%3CK, V%3E %7B%0A%0A  private RecordReader%3CK, V%3E reader;%0A%0A  public FilterRecordReader(RecordReader%3CK, V%3E reader) %7B%0A    this.reader = reader;%0A%0A  %7D%0A%0A  @Override%0A  public void close() throws IOException %7B%0A    reader.close();%0A  %7D%0A%0A  @Override%0A  public K getCurrentKey() throws IOException, InterruptedException %7B%0A    return reader.getCurrentKey();%0A  %7D%0A%0A  @Override%0A  public V getCurrentValue() throws IOException, InterruptedException %7B%0A    return reader.getCurrentValue();%0A  %7D%0A%0A  @Override%0A  public float getProgress() throws IOException, InterruptedException %7B%0A    return reader.getProgress();%0A  %7D%0A%0A  @Override%0A  public void initialize(InputSplit split, TaskAttemptContext context)%0A      throws IOException, InterruptedException %7B%0A    reader.initialize(split, context);%0A  %7D%0A%0A  @Override%0A  public boolean nextKeyValue() throws IOException, InterruptedException %7B%0A    return reader.nextKeyValue();%0A  %7D%0A%7D%0A

add FilterRecordReader which delegate all methods to a real record reader.