@Override
  public List<KinesisSource> split(int desiredNumSplits, PipelineOptions options) throws Exception {
    List<KinesisSource> sources = new ArrayList<>();
    KinesisReaderCheckpoint checkpoint;

    // in case split() is called upon existing checkpoints for further splitting:
    if (this.initialCheckpoint != null) {
      checkpoint = this.initialCheckpoint;
    }
    // in case a new checkpoint is created from scratch:
    else {
      AwsOptions awsOptions = options.as(AwsOptions.class);
      ClientConfiguration config = spec.getClientConfiguration();
      try (KinesisClient client = buildClient(awsOptions, KinesisClient.builder(), config)) {
        checkpoint = generateInitCheckpoint(spec, client);
      }
    }

    for (KinesisReaderCheckpoint partition : checkpoint.splitInto(desiredNumSplits)) {
      sources.add(new KinesisSource(spec, partition));
    }
    return sources;
  }
@Test
  public void testSplitGeneratesCorrectNumberOfSources() throws Exception {
    mockShards(kinesisClient, 3);
    KinesisSource source = sourceWithMockedKinesisClient(spec());
    assertThat(source.split(1, options).size()).isEqualTo(1);
    assertThat(source.split(2, options).size()).isEqualTo(2);
    assertThat(source.split(3, options).size()).isEqualTo(3);
    // there are only 3 shards, no more than 3 splits can be created
    assertThat(source.split(4, options).size()).isEqualTo(3);
  }