RecordOutputFormat extends FileOutputFormat { @Override public void configure(Configuration parameters) { super.configure(parameters); this.numFields = parameters.getInteger(NUM_FIELDS_PARAMETER, -1); if (this.numFields < 1) { throw new IllegalArgumentException("Invalid configuration for RecordOutputFormat: " + "Need to specify number of fields > 0."); } @SuppressWarnings("unchecked") Class<Value>[] arr = new Class[this.numFields]; this.classes = arr; for (int i = 0; i < this.numFields; i++) { @SuppressWarnings("unchecked") Class<? extends Value> clazz = (Class<? extends Value>) parameters.getClass(FIELD_TYPE_PARAMETER_PREFIX + i, null); if (clazz == null) { throw new IllegalArgumentException("Invalid configuration for RecordOutputFormat: " + "No type class for parameter " + i); } this.classes[i] = clazz; } this.recordPositions = new int[this.numFields]; boolean anyRecordPosDefined = false; boolean allRecordPosDefined = true; for(int i = 0; i < this.numFields; i++) { int pos = parameters.getInteger(RECORD_POSITION_PARAMETER_PREFIX + i, Integer.MIN_VALUE); if(pos != Integer.MIN_VALUE) { anyRecordPosDefined = true; if(pos < 0) { throw new IllegalArgumentException("Invalid configuration for RecordOutputFormat: " + "Invalid record position for parameter " + i); } this.recordPositions[i] = pos; } else { allRecordPosDefined = false; this.recordPositions[i] = i; } } if(anyRecordPosDefined && !allRecordPosDefined) { throw new IllegalArgumentException("Invalid configuration for RecordOutputFormat: " + "Either none or all record positions must be defined."); } this.recordDelimiter = parameters.getString(RECORD_DELIMITER_PARAMETER, AbstractConfigBuilder.NEWLINE_DELIMITER); if (this.recordDelimiter == null) { throw new IllegalArgumentException("The delimiter in the DelimitedOutputFormat must not be null."); } this.charsetName = parameters.getString(RECORD_DELIMITER_ENCODING, null); this.fieldDelimiter = parameters.getString(FIELD_DELIMITER_PARAMETER, "|"); this.lenient = parameters.getBoolean(LENIENT_PARSING, false); }  @Override void configure(Configuration parameters); @Override void open(int taskNumber); @Override void close(); @Override void writeRecord(PactRecord record); static ConfigBuilder configureRecordFormat(FileDataSink target); static final String RECORD_DELIMITER_PARAMETER; static final String FIELD_DELIMITER_PARAMETER; static final String NUM_FIELDS_PARAMETER; static final String FIELD_TYPE_PARAMETER_PREFIX; static final String RECORD_POSITION_PARAMETER_PREFIX; static final String LENIENT_PARSING; }
@Test public void testConfigure() { try { Configuration config = new Configuration(); config.setString(RecordOutputFormat.FILE_PARAMETER_KEY, "file: boolean validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setInteger(RecordOutputFormat.NUM_FIELDS_PARAMETER, 2); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 0, PactString.class); config.setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 1, PactInteger.class); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertTrue(validConfig); config.setInteger(RecordOutputFormat.NUM_FIELDS_PARAMETER, 3); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setClass(RecordOutputFormat.FIELD_TYPE_PARAMETER_PREFIX + 2, PactString.class); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertTrue(validConfig); config.setString(RecordOutputFormat.FIELD_DELIMITER_PARAMETER, "|"); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; System.out.println(iae.getMessage()); } assertTrue(validConfig); config.setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 1, 0); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertFalse(validConfig); config.setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 0, 3); config.setInteger(RecordOutputFormat.RECORD_POSITION_PARAMETER_PREFIX + 2, 9); validConfig = true; try { format.configure(config); } catch(IllegalArgumentException iae) { validConfig = false; } assertTrue(validConfig); } catch (Exception ex) { Assert.fail("Test failed due to a " + ex.getClass().getSimpleName() + ": " + ex.getMessage()); } }