@Override
  public UnboundedReader<KinesisRecord> createReader(
      PipelineOptions options, @Nullable KinesisReaderCheckpoint checkpointMark)
      throws IOException {
    KinesisReaderCheckpoint initCheckpoint;
    if (checkpointMark != null) {
      LOG.info("Got checkpoint mark {}", checkpointMark);
      initCheckpoint = checkpointMark;
    } else {
      try {
        LOG.info("No checkpointMark specified, fall back to initial {}", this.initialCheckpoint);
        initCheckpoint = Preconditions.checkArgumentNotNull(this.initialCheckpoint);
      } catch (Exception e) {
        throw new IOException(e);
      }
    }
    return initReader(spec, options, initCheckpoint, this);
  }
@Test
  public void testCreateReaderOfCorrectType() throws Exception {
    KinesisIO.Read readSpec =
        KinesisIO.read()
            .withStreamName("stream-xxx")
            .withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON);

    KinesisIO.Read readSpecEFO =
        KinesisIO.read()
            .withStreamName("stream-xxx")
            .withConsumerArn("consumer-aaa")
            .withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON);

    KinesisReaderCheckpoint initCheckpoint = new KinesisReaderCheckpoint(ImmutableList.of());

    UnboundedSource.UnboundedReader<KinesisRecord> reader =
        new KinesisSource(readSpec, initCheckpoint).createReader(options, null);
    assertThat(reader).isInstanceOf(KinesisReader.class);

    UnboundedSource.UnboundedReader<KinesisRecord> efoReader =
        new KinesisSource(readSpecEFO, initCheckpoint).createReader(options, null);
    assertThat(efoReader).isInstanceOf(EFOKinesisReader.class);
  }