// Path: src/main/java/com/alexholmes/hadooputils/combine/common/mapred/CommonCombineRecordReader.java
// public class CommonCombineRecordReader<K, V> implements RecordReader<K, V> {
//     private static final Log LOG = LogFactory.getLog(CommonCombineRecordReader.class);
// 
//     protected Configuration conf;
//     protected int currentSplit = -1;
//     protected RecordReader<K, V> reader;
//     protected CombineFileSplit split;
//     private final RecordReaderEngineerer<K, V> engineerer;
//     private long totalBytes;
// 
//     /**
//      * Create an instance of the class.
//      *
//      * @param conf  the Hadoop config
//      * @param split the input split
//      * @param engineerer the RecordReader engineering instance
//      * @throws IOException on io error
//      */
//     public CommonCombineRecordReader(Configuration conf, CombineFileSplit split, RecordReaderEngineerer<K, V> engineerer) throws IOException {
//         this.conf = conf;
// 
//         this.split = split;
//         this.engineerer = engineerer;
// 
//         for (int i = 0; i < this.split.getPaths().length; i++) {
//             totalBytes += this.split.getLength(i);
// 
//             if (LOG.isDebugEnabled()) {
//                 LOG.debug(String.format("Got split %s offest %d len %d", this.split.getPath(i), this.split.getOffset(i), this.split.getLength(i)));
//             }
//         }
// 
//         nextReader();
//     }
// 
//     /**
//      * Moves on to the next split inside {@link #split}. The {@link #reader} will be {@code null}
//      * once we have exhausted all the splits.
//      *
//      * @return true if we successfully moved on to the next split
//      * @throws java.io.IOException if we hit io errors
//      */
//     public boolean nextReader() throws IOException {
//         // close the current reader and set it to null
//         close();
// 
//         currentSplit++;
// 
//         if (currentSplit >= split.getPaths().length) {
//             // hit the end of the line
//             return false;
//         }
// 
//         FileSplit fileSplit = new FileSplit(
//                 split.getPath(currentSplit),
//                 split.getOffset(currentSplit),
//                 split.getLength(currentSplit),
//                 split.getLocations() == null || split.getLocations().length - 1 < currentSplit ? null : new String[]{split.getLocations()[currentSplit]});
// 
//         reader = engineerer.createRecordReader(conf, fileSplit);
//         return true;
//     }
// 
//     /**
//      * Return the progress within the input split
//      *
//      * @return 0.0 to 1.0 of the input byte range
//      */
//     @Override
//     public float getProgress() throws IOException {
//         if (reader == null) {
//             return 1.0f;
//         }
// 
//         long completeBytes = 0;
// 
//         for (int i = 0; i < currentSplit - 1; i++) {
//             completeBytes += split.getLength(i);
//         }
// 
//         float currentReaderProgress = reader.getProgress();
// 
//         completeBytes += Math.min(split.getLength(currentSplit), (long) ((float) split.getLength(currentSplit)) * currentReaderProgress);
// 
//         return (float) completeBytes / (float) totalBytes;
//     }
// 
//     @Override
//     public boolean next(K key, V value) throws IOException {
// 
//         while (reader != null) {
//             if (reader.next(key, value)) {
//                 return true;
//             }
//             nextReader();
//         }
//         return false;
//     }
// 
//     @Override
//     public K createKey() {
//         return reader.createKey();
//     }
// 
//     @Override
//     public V createValue() {
//         return reader.createValue();
//     }
// 
//     @Override
//     public long getPos() throws IOException {
//         return (long) (((float) totalBytes) * getProgress());
//     }
// 
//     @Override
//     public synchronized void close() throws IOException {
//         if (reader != null) {
//             reader.close();
//             reader = null;
//         }
//     }
// 
//     /**
//      * Create {@link RecordReader} instances.
//      *
//      * @param <K> the {@link RecordReader} key type
//      * @param <V> the {@link RecordReader} value type
//      */
//     public static interface RecordReaderEngineerer<K, V> {
//         /**
//          * Create an instance of a {@link RecordReader}.
//          *
//          * @param conf the Hadoop conf
//          * @param split the input split
//          * @return the RR
//          */
//         public RecordReader<K, V> createRecordReader(Configuration conf, FileSplit split) throws IOException;
//     }
// }


import static org.junit.Assert.*;
import com.alexholmes.hadooputils.combine.common.mapred.CommonCombineRecordReader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.mapred.*;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.IOException;

        SequenceFile.Writer writer =
                SequenceFile.createWriter(fs, conf, path, Text.class,
                        Text.class,
                        SequenceFile.CompressionType.BLOCK,
                        new DefaultCodec());
        try {
            writer.append(key, value);
        } finally {
            writer.close();
        }
    }

    @Test
    public void testOneFile() throws IOException, InterruptedException {
        Path dir = new Path(tempFolder.getRoot().getAbsolutePath());

        CombineSequenceFileInputFormat<Text, Text> inputFormat = new CombineSequenceFileInputFormat<Text, Text>();
        Path inputFile = new Path(dir, "file1.txt");

        writeSequenceFile(inputFile);

        Configuration conf = new Configuration();
        JobConf jobConf = new JobConf(conf);

        FileInputFormat.addInputPath(jobConf, inputFile);

        InputSplit[] splits = inputFormat.getSplits(jobConf, 1);
        assertEquals(1, splits.length);

// Path: src/main/java/com/alexholmes/hadooputils/combine/common/mapred/CommonCombineRecordReader.java
// public class CommonCombineRecordReader<K, V> implements RecordReader<K, V> {
//     private static final Log LOG = LogFactory.getLog(CommonCombineRecordReader.class);
// 
//     protected Configuration conf;
//     protected int currentSplit = -1;
//     protected RecordReader<K, V> reader;
//     protected CombineFileSplit split;
//     private final RecordReaderEngineerer<K, V> engineerer;
//     private long totalBytes;
// 
//     /**
//      * Create an instance of the class.
//      *
//      * @param conf  the Hadoop config
//      * @param split the input split
//      * @param engineerer the RecordReader engineering instance
//      * @throws IOException on io error
//      */
//     public CommonCombineRecordReader(Configuration conf, CombineFileSplit split, RecordReaderEngineerer<K, V> engineerer) throws IOException {
//         this.conf = conf;
// 
//         this.split = split;
//         this.engineerer = engineerer;
// 
//         for (int i = 0; i < this.split.getPaths().length; i++) {
//             totalBytes += this.split.getLength(i);
// 
//             if (LOG.isDebugEnabled()) {
//                 LOG.debug(String.format("Got split %s offest %d len %d", this.split.getPath(i), this.split.getOffset(i), this.split.getLength(i)));
//             }
//         }
// 
//         nextReader();
//     }
// 
//     /**
//      * Moves on to the next split inside {@link #split}. The {@link #reader} will be {@code null}
//      * once we have exhausted all the splits.
//      *
//      * @return true if we successfully moved on to the next split
//      * @throws java.io.IOException if we hit io errors
//      */
//     public boolean nextReader() throws IOException {
//         // close the current reader and set it to null
//         close();
// 
//         currentSplit++;
// 
//         if (currentSplit >= split.getPaths().length) {
//             // hit the end of the line
//             return false;
//         }
// 
//         FileSplit fileSplit = new FileSplit(
//                 split.getPath(currentSplit),
//                 split.getOffset(currentSplit),
//                 split.getLength(currentSplit),
//                 split.getLocations() == null || split.getLocations().length - 1 < currentSplit ? null : new String[]{split.getLocations()[currentSplit]});
// 
//         reader = engineerer.createRecordReader(conf, fileSplit);
//         return true;
//     }
// 
//     /**
//      * Return the progress within the input split
//      *
//      * @return 0.0 to 1.0 of the input byte range
//      */
//     @Override
//     public float getProgress() throws IOException {
//         if (reader == null) {
//             return 1.0f;
//         }
// 
//         long completeBytes = 0;
// 
//         for (int i = 0; i < currentSplit - 1; i++) {
//             completeBytes += split.getLength(i);
//         }
// 
//         float currentReaderProgress = reader.getProgress();
// 
//         completeBytes += Math.min(split.getLength(currentSplit), (long) ((float) split.getLength(currentSplit)) * currentReaderProgress);
// 
//         return (float) completeBytes / (float) totalBytes;
//     }
// 
//     @Override
//     public boolean next(K key, V value) throws IOException {
// 
//         while (reader != null) {
//             if (reader.next(key, value)) {
//                 return true;
//             }
//             nextReader();
//         }
//         return false;
//     }
// 
//     @Override
//     public K createKey() {
//         return reader.createKey();
//     }
// 
//     @Override
//     public V createValue() {
//         return reader.createValue();
//     }
// 
//     @Override
//     public long getPos() throws IOException {
//         return (long) (((float) totalBytes) * getProgress());
//     }
// 
//     @Override
//     public synchronized void close() throws IOException {
//         if (reader != null) {
//             reader.close();
//             reader = null;
//         }
//     }
// 
//     /**
//      * Create {@link RecordReader} instances.
//      *
//      * @param <K> the {@link RecordReader} key type
//      * @param <V> the {@link RecordReader} value type
//      */
//     public static interface RecordReaderEngineerer<K, V> {
//         /**
//          * Create an instance of a {@link RecordReader}.
//          *
//          * @param conf the Hadoop conf
//          * @param split the input split
//          * @return the RR
//          */
//         public RecordReader<K, V> createRecordReader(Configuration conf, FileSplit split) throws IOException;
//     }
// }


// Path: src/test/java/com/alexholmes/hadooputils/combine/seqfile/mapred/CombineSequenceFileTest.java
import static org.junit.Assert.*;
import com.alexholmes.hadooputils.combine.common.mapred.CommonCombineRecordReader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.mapred.*;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.IOException;


        SequenceFile.Writer writer =
                SequenceFile.createWriter(fs, conf, path, Text.class,
                        Text.class,
                        SequenceFile.CompressionType.BLOCK,
                        new DefaultCodec());
        try {
            writer.append(key, value);
        } finally {
            writer.close();
        }
    }

    @Test
    public void testOneFile() throws IOException, InterruptedException {
        Path dir = new Path(tempFolder.getRoot().getAbsolutePath());

        CombineSequenceFileInputFormat<Text, Text> inputFormat = new CombineSequenceFileInputFormat<Text, Text>();
        Path inputFile = new Path(dir, "file1.txt");

        writeSequenceFile(inputFile);

        Configuration conf = new Configuration();
        JobConf jobConf = new JobConf(conf);

        FileInputFormat.addInputPath(jobConf, inputFile);

        InputSplit[] splits = inputFormat.getSplits(jobConf, 1);
        assertEquals(1, splits.length);

        CommonCombineRecordReader<Text, Text> rr = (CommonCombineRecordReader<Text, Text>) inputFormat.getRecordReader(splits[0], jobConf, new DummyReporter());