class AvroRecordReader extends RecordReader<LongWritable, GenericData.Record> {

  protected void initializeInternal(InputSplit inputSplit, Configuration conf) throws IOException {
    Preconditions.checkState(
        inputSplit instanceof FileSplit, "AvroRecordReader requires FileSplit input splits.");
    FileSplit fileSplit = (FileSplit) inputSplit;
    splitStart = fileSplit.getStart();
    splitLength = fileSplit.getLength();

    Path filePath = fileSplit.getPath();
    FileSystem fs = filePath.getFileSystem(conf);
    FileStatus status = fs.getFileStatus(filePath);
    inputFileLength = status.getLen();

    final FSDataInputStream stream = fs.open(filePath);
    dataFileReader =
        DataFileReader.openReader(
            new SeekableInput() {
              @Override
              public void seek(long offset) throws IOException {
                stream.seek(offset);
              }

              @Override
              public long tell() throws IOException {
                return stream.getPos();
              }

              @Override
              public long length() {
                return inputFileLength;
              }

              @Override
              public int read(byte[] bytes, int offset, int length) throws IOException {
                return stream.read(bytes, offset, length);
              }

              @Override
              public void close() throws IOException {
                stream.close();
              }
            },
            new GenericDatumReader<GenericData.Record>());
    // Sync to the first sync point after the start of the split:
    dataFileReader.sync(fileSplit.getStart());
    schema = dataFileReader.getSchema();
    currentRecord = new GenericData.Record(schema);
  }

}