EmptyStringRecordFilter implements RecordFilter<String> { @Override public Record<String> processRecord(Record<String> record) { String payload = record.getPayload(); if (payload.isEmpty()) { return null; } return record; }  @Override Record<String> processRecord(Record<String> record);  }
@Test public void emptyRecordsShouldBeFiltered() { when(record.getPayload()).thenReturn(""); assertThat(filter.processRecord(record)).isNull(); }
@Test public void nonEmptyRecordsShouldNotBeFiltered() { when(record.getPayload()).thenReturn("foo"); assertThat(filter.processRecord(record)).isEqualTo(record); }