SimplifiedKinesisClient { public String getShardIterator( final String streamName, final String shardId, final ShardIteratorType shardIteratorType, final String startingSequenceNumber, final Instant timestamp) throws TransientKinesisException { return wrapExceptions( () -> kinesis .getShardIterator( GetShardIteratorRequest.builder() .streamName(streamName) .shardId(shardId) .shardIteratorType(shardIteratorType) .startingSequenceNumber(startingSequenceNumber) .timestamp(TimeUtil.toJava(timestamp)) .build()) .shardIterator()); } SimplifiedKinesisClient(
      KinesisClient kinesis, CloudWatchClient cloudWatch, Integer limit); static SimplifiedKinesisClient from(AWSClientsProvider provider, Integer limit); String getShardIterator(
      final String streamName,
      final String shardId,
      final ShardIteratorType shardIteratorType,
      final String startingSequenceNumber,
      final Instant timestamp); List<Shard> listShards(final String streamName); GetKinesisRecordsResult getRecords(String shardIterator, String streamName, String shardId); GetKinesisRecordsResult getRecords(
      final String shardIterator,
      final String streamName,
      final String shardId,
      final Integer limit); static List<KinesisClientRecord> deaggregate(List<Record> records); long getBacklogBytes(String streamName, Instant countSince); long getBacklogBytes(
      final String streamName, final Instant countSince, final Instant countTo);  }
@Test public void shouldReturnIteratorStartingWithSequenceNumber() throws Exception { when(kinesis.getShardIterator( GetShardIteratorRequest.builder() .streamName(STREAM) .shardId(SHARD_1) .shardIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER) .startingSequenceNumber(SEQUENCE_NUMBER) .build())) .thenReturn(GetShardIteratorResponse.builder().shardIterator(SHARD_ITERATOR).build()); String stream = underTest.getShardIterator( STREAM, SHARD_1, ShardIteratorType.AT_SEQUENCE_NUMBER, SEQUENCE_NUMBER, null); assertThat(stream).isEqualTo(SHARD_ITERATOR); }
@Test public void shouldReturnIteratorStartingWithTimestamp() throws Exception { Instant timestamp = Instant.now(); when(kinesis.getShardIterator( GetShardIteratorRequest.builder() .streamName(STREAM) .shardId(SHARD_1) .shardIteratorType(ShardIteratorType.AT_SEQUENCE_NUMBER) .timestamp(TimeUtil.toJava(timestamp)) .build())) .thenReturn(GetShardIteratorResponse.builder().shardIterator(SHARD_ITERATOR).build()); String stream = underTest.getShardIterator( STREAM, SHARD_1, ShardIteratorType.AT_SEQUENCE_NUMBER, null, timestamp); assertThat(stream).isEqualTo(SHARD_ITERATOR); }