class 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;
  }

}