DelimitedInputFormat extends FileInputFormat { @Override public void configure(Configuration parameters) { super.configure(parameters); final String delimString = parameters.getString(RECORD_DELIMITER, AbstractConfigBuilder.NEWLINE_DELIMITER); if (delimString == null) { throw new IllegalArgumentException("The delimiter not be null."); } final String charsetName = parameters.getString(RECORD_DELIMITER_ENCODING, null); try { this.delimiter = charsetName == null ? delimString.getBytes() : delimString.getBytes(charsetName); } catch (UnsupportedEncodingException useex) { throw new IllegalArgumentException("The charset with the name '" + charsetName + "' is not supported on this TaskManager instance.", useex); } this.numLineSamples = DEFAULT_NUM_SAMPLES; final String samplesString = parameters.getString(NUM_STATISTICS_SAMPLES, null); if (samplesString != null) { try { this.numLineSamples = Integer.parseInt(samplesString); } catch (NumberFormatException nfex) { if (LOG.isWarnEnabled()) LOG.warn("Invalid value for number of samples to take: " + samplesString + ". Using default value of " + DEFAULT_NUM_SAMPLES); } } }  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 testConfigure() { when(this.config.getString(Matchers.matches(DelimitedInputFormat.RECORD_DELIMITER), Matchers.anyString())) .thenReturn("\n"); when(this.config.getString(Matchers.matches(FileInputFormat.FILE_PARAMETER_KEY), Matchers.anyString())) .thenReturn("file: format.configure(this.config); verify(this.config, times(4)).getString(Matchers.any(String.class), Matchers.any(String.class)); assertEquals("\n", new String(format.getDelimiter())); when(this.config.getString(Matchers.matches(DelimitedInputFormat.RECORD_DELIMITER), Matchers.anyString())) .thenReturn("&-&"); format.configure(this.config); verify(this.config, times(8)).getString(Matchers.any(String.class), Matchers.any(String.class)); assertEquals("&-&", new String(format.getDelimiter())); }