RecordFilter { public List<KinesisRecord> apply(List<KinesisRecord> records, ShardCheckpoint checkpoint) { List<KinesisRecord> filteredRecords = newArrayList(); for (KinesisRecord record : records) { if (checkpoint.isBeforeOrAt(record)) { filteredRecords.add(record); } } return filteredRecords; }  List<KinesisRecord> apply(List<KinesisRecord> records, ShardCheckpoint checkpoint);  }
@Test public void shouldFilterOutRecordsBeforeOrAtCheckpoint() { when(checkpoint.isBeforeOrAt(record1)).thenReturn(false); when(checkpoint.isBeforeOrAt(record2)).thenReturn(true); when(checkpoint.isBeforeOrAt(record3)).thenReturn(true); when(checkpoint.isBeforeOrAt(record4)).thenReturn(false); when(checkpoint.isBeforeOrAt(record5)).thenReturn(true); List<KinesisRecord> records = Lists.newArrayList(record1, record2, record3, record4, record5); RecordFilter underTest = new RecordFilter(); List<KinesisRecord> retainedRecords = underTest.apply(records, checkpoint); Assertions.assertThat(retainedRecords).containsOnly(record2, record3, record5); }
@Test public void shouldNotFailOnEmptyList() { List<KinesisRecord> records = Collections.emptyList(); RecordFilter underTest = new RecordFilter(); List<KinesisRecord> retainedRecords = underTest.apply(records, checkpoint); Assertions.assertThat(retainedRecords).isEmpty(); }