@Test public void testTimeoutExceptionIsRetryableForGetShardIterator() throws Exception { expectedExceptionRule.expect(RetryableRetrievalException.class); expectedExceptionRule.expectCause(isA(TimeoutException.class)); expectedExceptionRule.expectMessage("Timeout"); when(kinesisClient.getShardIterator(any(GetShardIteratorRequest.class))) .thenReturn(getShardIteratorResponseFuture); when(getShardIteratorResponseFuture.get(anyLong(), any(TimeUnit.class))) .thenThrow(new TimeoutException("Timeout")); kinesisDataFetcher.initialize(SentinelCheckpoint.LATEST.toString(), INITIAL_POSITION_LATEST); }
@Override public void initialize(final String initialCheckpoint, final InitialPositionInStreamExtended initialPositionInStream) { log.info("Initializing shard {} with {}", streamAndShardId, initialCheckpoint); advanceIteratorTo(initialCheckpoint, initialPositionInStream); isInitialized = true; }
KinesisDataFetcher implements DataFetcher { @Override public void initialize(final String initialCheckpoint, final InitialPositionInStreamExtended initialPositionInStream) { log.info("Initializing shard {} with {}", streamAndShardId, initialCheckpoint); advanceIteratorTo(initialCheckpoint, initialPositionInStream); isInitialized = true; } }
KinesisDataFetcher implements DataFetcher { @Override public void initialize(final String initialCheckpoint, final InitialPositionInStreamExtended initialPositionInStream) { log.info("Initializing shard {} with {}", streamAndShardId, initialCheckpoint); advanceIteratorTo(initialCheckpoint, initialPositionInStream); isInitialized = true; } @Deprecated  KinesisDataFetcher(KinesisAsyncClient kinesisClient, String streamName, String shardId, int maxRecords, MetricsFactory metricsFactory);  KinesisDataFetcher(KinesisAsyncClient kinesisClient, DataFetcherProviderConfig kinesisDataFetcherProviderConfig); }
KinesisDataFetcher implements DataFetcher { @Override public void initialize(final String initialCheckpoint, final InitialPositionInStreamExtended initialPositionInStream) { log.info("Initializing shard {} with {}", streamAndShardId, initialCheckpoint); advanceIteratorTo(initialCheckpoint, initialPositionInStream); isInitialized = true; } @Deprecated  KinesisDataFetcher(KinesisAsyncClient kinesisClient, String streamName, String shardId, int maxRecords, MetricsFactory metricsFactory);  KinesisDataFetcher(KinesisAsyncClient kinesisClient, DataFetcherProviderConfig kinesisDataFetcherProviderConfig); @Override DataFetcherResult getRecords(); @Override void initialize(final String initialCheckpoint,
                           final InitialPositionInStreamExtended initialPositionInStream); @Override void initialize(final ExtendedSequenceNumber initialCheckpoint,
                           final InitialPositionInStreamExtended initialPositionInStream); @Override void advanceIteratorTo(final String sequenceNumber,
                                  final InitialPositionInStreamExtended initialPositionInStream); @Override void restartIterator(); @Override void resetIterator(String shardIterator, String sequenceNumber, InitialPositionInStreamExtended initialPositionInStream); @Override GetRecordsResponse getGetRecordsResponse(GetRecordsRequest request); @Override GetRecordsRequest getGetRecordsRequest(String nextIterator); @Override String getNextIterator(GetShardIteratorRequest request); @Override GetRecordsResponse getRecords(@NonNull final String nextIterator); }
KinesisDataFetcher implements DataFetcher { @Override public void initialize(final String initialCheckpoint, final InitialPositionInStreamExtended initialPositionInStream) { log.info("Initializing shard {} with {}", streamAndShardId, initialCheckpoint); advanceIteratorTo(initialCheckpoint, initialPositionInStream); isInitialized = true; } @Deprecated  KinesisDataFetcher(KinesisAsyncClient kinesisClient, String streamName, String shardId, int maxRecords, MetricsFactory metricsFactory);  KinesisDataFetcher(KinesisAsyncClient kinesisClient, DataFetcherProviderConfig kinesisDataFetcherProviderConfig); @Override DataFetcherResult getRecords(); @Override void initialize(final String initialCheckpoint,
                           final InitialPositionInStreamExtended initialPositionInStream); @Override void initialize(final ExtendedSequenceNumber initialCheckpoint,
                           final InitialPositionInStreamExtended initialPositionInStream); @Override void advanceIteratorTo(final String sequenceNumber,
                                  final InitialPositionInStreamExtended initialPositionInStream); @Override void restartIterator(); @Override void resetIterator(String shardIterator, String sequenceNumber, InitialPositionInStreamExtended initialPositionInStream); @Override GetRecordsResponse getGetRecordsResponse(GetRecordsRequest request); @Override GetRecordsRequest getGetRecordsRequest(String nextIterator); @Override String getNextIterator(GetShardIteratorRequest request); @Override GetRecordsResponse getRecords(@NonNull final String nextIterator);  }