LineRegexReplaceInRegionBolt extends AbstractLogRecordBolt { void populateLogRecord(LogRecord logRecord, String record) { try (CSVReader csvReader = new CSVReader(new StringReader(record), delimiter)) { String[] values = csvReader.readNext(); int fieldSize = Math.min(fieldList.size(), values.length); for (int i = 0; i < fieldSize; i++) { String field = fieldList.get(i); String value = values[i]; logRecord.setValue(field, value); } } catch (IOException e) { logger.error("Failed to parse line: {}", record); throw new FailedException(e); } }  @Override void configure(Configuration conf);  }
@Test public void testPopulateLogRecord() throws Exception { String record = getResourceAsString(this.getClass(), "testPopulateLogRecord.txt"); bolt.delimiter = '\t'; bolt.fieldList = Arrays.asList( "date", "time", "src_user", "src_ip", "cs_method", "cs_uri_stem", "protocol", "r_ip", "http_response", "rs_bytes", "time_taken", "s_ip", "s_action", "bytes_out", "bytes_in" ); LogRecord logRecord = new LogRecord(); bolt.populateLogRecord(logRecord, record); assertThat(logRecord.getValue("date"), is("2015-01-02")); assertThat(logRecord.getValue("r_ip"), is("1.2.3.4")); }