@Test public void getRecords_GivenStreamWithMultipleShards_ShouldRequestRecordsForEachShard() throws IOException { Shard[] shards = new Shard[] { Shard.builder().shardId(UUID.randomUUID().toString()).build(), Shard.builder().shardId(UUID.randomUUID().toString()).build(), Shard.builder().shardId(UUID.randomUUID().toString()).build() }; mockDescribeStream(shards); String[] iterators = new String[3]; for (int i = 0; i < shards.length; i++) { iterators[i] = mockGetShardIterator(shards[i].shardId()); List<SampleRecord> records = Collections.singletonList(new SampleRecord()); mockGetRecords(iterators[i], null, records); } reader.getRecords(); Mockito.verify(mockClient).describeStream(describeStreamRequest); for (int i = 0; i < shards.length; i++) { verifyGetShardIterator(shards[i].shardId()); verifyGetRecords(iterators[i]); } }
@Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } KinesisStreamReader(String streamName, AbstractKinesisRecordTranslator<T> recordTranslator);  KinesisStreamReader(KinesisClient client,
                               String streamName,
                               AbstractKinesisRecordTranslator<T> recordTranslator); }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } KinesisStreamReader(String streamName, AbstractKinesisRecordTranslator<T> recordTranslator);  KinesisStreamReader(KinesisClient client,
                               String streamName,
                               AbstractKinesisRecordTranslator<T> recordTranslator); @Override List<T> getRecords(); @Override void setConfiguration(ReaderConfiguration configuration); }
KinesisStreamReader implements Reader<T> { @Override public List<T> getRecords() throws IOException { List<Shard> shards = this.client .describeStream(DescribeStreamRequest.builder().streamName(streamName).build()) .streamDescription().shards(); List<T> items = new LinkedList<>(); for (Shard shard : shards) { items.addAll(recordTranslator.toValues(getRecordsForShard(shard))); } return items; } KinesisStreamReader(String streamName, AbstractKinesisRecordTranslator<T> recordTranslator);  KinesisStreamReader(KinesisClient client,
                               String streamName,
                               AbstractKinesisRecordTranslator<T> recordTranslator); @Override List<T> getRecords(); @Override void setConfiguration(ReaderConfiguration configuration);  }