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;
  }
@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);
  }