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 shouldNotFailOnEmptyList() {
    List<KinesisRecord> records = Collections.emptyList();
    RecordFilter underTest = new RecordFilter();

    List<KinesisRecord> retainedRecords = underTest.apply(records, checkpoint);

    Assertions.assertThat(retainedRecords).isEmpty();
  }