LineRegexReplaceInRegionBolt extends AbstractLogRecordBolt { String replace(String record, Pattern pattern, String regex, String replacement) { Matcher match = pattern.matcher(record); if (match.find() && match.groupCount() > 0) { int startPos = match.start(1); int stopPos = match.end(1); String replaceString = match.group(1).replaceAll(regex, replacement); return record.substring(0, startPos) + replaceString + record.substring(stopPos); } else { return record; } }  @Override void configure(Configuration conf);  }
@Test public void testReplace() throws Exception { String record = getResourceAsString(this.getClass(), "testReplace.txt"); Pattern pattern = Pattern.compile("^([^\"]+)"); String regex = "[ ]"; String replacement = "|"; String result = bolt.replace(record, pattern, regex, replacement); assertThat(result, is(getResourceAsString(this.getClass(), "testReplaceResult.txt"))); }