DelimitedInputFormat extends FileInputFormat { @Override public void open(FileInputSplit split) throws IOException { super.open(split); this.bufferSize = this.bufferSize <= 0 ? DEFAULT_READ_BUFFER_SIZE : this.bufferSize; this.readBuffer = new byte[this.bufferSize]; this.wrapBuffer = new byte[256]; this.readPos = 0; this.overLimit = false; this.end = false; if (this.splitStart != 0) { this.stream.seek(this.splitStart); readLine(); if (this.overLimit) { this.end = true; } } else { fillBuffer(); } }  abstract boolean readRecord(PactRecord target, byte[] bytes, int offset, int numBytes); byte[] getDelimiter(); void setBufferSize(int bufferSize); int getBufferSize(); @Override void configure(Configuration parameters); @Override FileBaseStatistics getStatistics(BaseStatistics cachedStatistics); @Override void open(FileInputSplit split); @Override boolean reachedEnd(); @Override boolean nextRecord(PactRecord record); @Override void close(); static ConfigBuilder configureDelimitedFormat(FileDataSource target); static final String RECORD_DELIMITER; }
@Test public void testOpen() throws IOException { final String myString = "my mocked line 1\nmy mocked line 2\n"; final FileInputSplit split = createTempFile(myString); int bufferSize = 5; format.setBufferSize(bufferSize); format.open(split); assertEquals(0, format.splitStart); assertEquals(myString.length() - bufferSize, format.splitLength); assertEquals(bufferSize, format.getBufferSize()); }