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>()); dataFileReader.sync(fileSplit.getStart()); schema = dataFileReader.getSchema(); currentRecord = new GenericData.Record(schema); }  @Override void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext); @Override boolean nextKeyValue(); @Override LongWritable getCurrentKey(); @Override GenericData.Record getCurrentValue(); @Override float getProgress(); @Override void close();  }
@Test public void testMultipleSplits() throws IOException { long fileLength = testAvroFile.length(); List<FileSplit> splits = new ArrayList<>(); Path hadoopPath = new Path("file", null, testAvroFile.getAbsolutePath()); for (int blockStart = 0; blockStart < fileLength; blockStart += AUTO_SYNC_INTERVAL) { splits.add(new FileSplit(hadoopPath, blockStart, AUTO_SYNC_INTERVAL, new String[0])); } List<String> allRecordKeys = new ArrayList<>(); long totalFileRecords = 0; for (FileSplit split : splits) { try (AvroRecordReader reader = new AvroRecordReader()) { reader.initializeInternal(split, new Configuration()); List<String> keysInSplit = collectRecordKeys(reader); allRecordKeys.addAll(keysInSplit); int recordsInSplit = keysInSplit.size(); totalFileRecords += recordsInSplit; Truth.assertThat(recordsInSplit) .isLessThan(RECORD_COUNT); } } Truth.assertThat(allRecordKeys).containsExactlyElementsIn(allAddedKeys); Truth.assertThat(totalFileRecords).isEqualTo(RECORD_COUNT); }