ShardReadersPool { void start() throws TransientKinesisException { ImmutableMap.Builder<String, ShardRecordsIterator> shardsMap = ImmutableMap.builder(); for (ShardCheckpoint checkpoint : initialCheckpoint) { shardsMap.put(checkpoint.getShardId(), createShardIterator(kinesis, checkpoint)); } shardIteratorsMap.set(shardsMap.build()); if (!shardIteratorsMap.get().isEmpty()) { recordsQueue = new ArrayBlockingQueue<>(queueCapacityPerShard * shardIteratorsMap.get().size()); startReadingShards(shardIteratorsMap.get().values()); } else { recordsQueue = new ArrayBlockingQueue<>(1); } } ShardReadersPool(
      SimplifiedKinesisClient kinesis,
      KinesisReaderCheckpoint initialCheckpoint,
      WatermarkPolicyFactory watermarkPolicyFactory,
      RateLimitPolicyFactory rateLimitPolicyFactory,
      int queueCapacityPerShard);  static final int DEFAULT_CAPACITY_PER_SHARD; }
@Test public void shouldStopReadingShardAfterReceivingShardClosedException() throws Exception { when(firstIterator.readNextBatch()).thenThrow(KinesisShardClosedException.class); when(firstIterator.findSuccessiveShardRecordIterators()).thenReturn(Collections.emptyList()); shardReadersPool.start(); verify(firstIterator, timeout(TIMEOUT_IN_MILLIS).times(1)).readNextBatch(); verify(secondIterator, timeout(TIMEOUT_IN_MILLIS).atLeast(2)).readNextBatch(); }
@Test public void shouldStartReadingSuccessiveShardsAfterReceivingShardClosedException() throws Exception { when(firstIterator.readNextBatch()).thenThrow(KinesisShardClosedException.class); when(firstIterator.findSuccessiveShardRecordIterators()) .thenReturn(ImmutableList.of(thirdIterator, fourthIterator)); shardReadersPool.start(); verify(thirdIterator, timeout(TIMEOUT_IN_MILLIS).atLeast(2)).readNextBatch(); verify(fourthIterator, timeout(TIMEOUT_IN_MILLIS).atLeast(2)).readNextBatch(); }
@Test public void shouldStopReadersPoolWhenLastShardReaderStopped() throws Exception { when(firstIterator.readNextBatch()).thenThrow(KinesisShardClosedException.class); when(firstIterator.findSuccessiveShardRecordIterators()).thenReturn(Collections.emptyList()); shardReadersPool.start(); verify(firstIterator, timeout(TIMEOUT_IN_MILLIS).times(1)).readNextBatch(); }
@Test public void shouldStopReadersPoolAlsoWhenExceptionsOccurDuringStopping() throws Exception { when(firstIterator.readNextBatch()).thenThrow(KinesisShardClosedException.class); when(firstIterator.findSuccessiveShardRecordIterators()) .thenThrow(TransientKinesisException.class) .thenReturn(Collections.emptyList()); shardReadersPool.start(); verify(firstIterator, timeout(TIMEOUT_IN_MILLIS).times(2)).readNextBatch(); }