void start(Iterable<ShardCheckpoint> checkpoints) {
    LOG.info(
        "Pool {} - starting for stream {} consumer {}. Checkpoints = {}",
        poolId,
        read.getStreamName(),
        consumerArn,
        checkpoints);
    for (ShardCheckpoint shardCheckpoint : checkpoints) {
      checkState(
          !state.containsKey(shardCheckpoint.getShardId()),
          "Duplicate shard id %s",
          shardCheckpoint.getShardId());
      ShardState shardState =
          new ShardState(
              initShardSubscriber(shardCheckpoint), shardCheckpoint, watermarkPolicyFactory);
      state.put(shardCheckpoint.getShardId(), shardState);
    }
  }
@Test
  public void skipsEntireAggregatedBatch() throws Exception {
    kinesis = new EFOStubbedKinesisAsyncClient(10);
    SubscribeToShardEvent eventWithAggRecords = eventWithAggRecords(12, 6);
    kinesis.stubSubscribeToShard("shard-000", eventWithAggRecords);

    KinesisReaderCheckpoint initialCheckpoint =
        new KinesisReaderCheckpoint(
            ImmutableList.of(
                new ShardCheckpoint(
                    STREAM, "shard-000", ShardIteratorType.AFTER_SEQUENCE_NUMBER, "12", 5L)));
    pool = new EFOShardSubscribersPool(readSpec, consumerArn, kinesis);
    pool.start(initialCheckpoint);

    List<KinesisRecord> actualRecords = waitForRecords(pool, 10);
    assertThat(actualRecords.size()).isEqualTo(0);
    assertThat(kinesis.subscribeRequestsSeen())
        .containsExactlyInAnyOrder(
            subscribeAtSeqNumber("shard-000", "12"),
            // to no-op subscriber
            subscribeAfterSeqNumber("shard-000", "12"));
  }