KinesisReader extends UnboundedSource.UnboundedReader<KinesisRecord> { @Override public boolean start() throws IOException { LOG.info("Starting reader using {}", initialCheckpointGenerator); try { shardReadersPool = createShardReadersPool(); shardReadersPool.start(); } catch (TransientKinesisException e) { throw new IOException(e); } return advance(); } KinesisReader(
      SimplifiedKinesisClient kinesis,
      CheckpointGenerator initialCheckpointGenerator,
      KinesisSource source,
      WatermarkPolicyFactory watermarkPolicyFactory,
      RateLimitPolicyFactory rateLimitPolicyFactory,
      Duration upToDateThreshold,
      Integer maxCapacityPerShard);   KinesisReader(
      SimplifiedKinesisClient kinesis,
      CheckpointGenerator initialCheckpointGenerator,
      KinesisSource source,
      WatermarkPolicyFactory watermarkPolicyFactory,
      RateLimitPolicyFactory rateLimitPolicyFactory,
      Duration upToDateThreshold,
      Duration backlogBytesCheckThreshold,
      Integer maxCapacityPerShard); @Override boolean start(); @Override boolean advance(); @Override byte[] getCurrentRecordId(); @Override KinesisRecord getCurrent(); @Override Instant getCurrentTimestamp(); @Override void close(); @Override Instant getWatermark(); @Override UnboundedSource.CheckpointMark getCheckpointMark(); @Override UnboundedSource<KinesisRecord, ?> getCurrentSource(); @Override long getSplitBacklogBytes();  }
@Test public void startReturnsFalseIfNoDataAtTheBeginning() throws IOException { assertThat(reader.start()).isFalse(); }
@Test public void startReturnsTrueIfSomeDataAvailable() throws IOException { when(shardReadersPool.nextRecord()) .thenReturn(CustomOptional.of(a)) .thenReturn(CustomOptional.absent()); assertThat(reader.start()).isTrue(); }