GrepFilter implements RecordFilter<String> { @Override public Record<String> processRecord(Record<String> record) { String payload = record.getPayload(); if (!payload.contains(pattern)) { return null; } return record; } GrepFilter(final String pattern); @Override Record<String> processRecord(Record<String> record);  }
@Test public void whenRecordContainsPattern_ThenItShouldNotBeFiltered() { when(record.getPayload()).thenReturn("java rocks!"); assertThat(grepFilter.processRecord(record)).isEqualTo(record); }
@Test public void whenRecordDoesNotContainPattern_ThenItShouldBeFiltered() { when(record.getPayload()).thenReturn("c++ .."); assertThat(grepFilter.processRecord(record)).isNull(); }
@Test public void patternLookupShouldBeCaseSensitive() { when(record.getPayload()).thenReturn("JAVA rocks!"); assertThat(grepFilter.processRecord(record)).isNull(); }