@Test public void testadvanceIteratorTo() throws KinesisClientLibException, InterruptedException, ExecutionException { final Checkpointer checkpoint = mock(Checkpointer.class); final String iteratorA = "foo"; final String iteratorB = "bar"; final String seqA = "123"; final String seqB = "456"; ArgumentCaptor<GetShardIteratorRequest> shardIteratorRequestCaptor = ArgumentCaptor .forClass(GetShardIteratorRequest.class); when(kinesisClient.getShardIterator(shardIteratorRequestCaptor.capture())) .thenReturn(makeGetShardIteratorResonse(iteratorA)).thenReturn(makeGetShardIteratorResonse(iteratorA)) .thenReturn(makeGetShardIteratorResonse(iteratorB)); when(checkpoint.getCheckpoint(SHARD_ID)).thenReturn(new ExtendedSequenceNumber(seqA)); kinesisDataFetcher.initialize(seqA, null); kinesisDataFetcher.advanceIteratorTo(seqA, null); kinesisDataFetcher.advanceIteratorTo(seqB, null); final List<GetShardIteratorRequest> shardIteratorRequests = shardIteratorRequestCaptor.getAllValues(); assertEquals(3, shardIteratorRequests.size()); int count = 0; for (GetShardIteratorRequest request : shardIteratorRequests) { assertEquals(STREAM_NAME, request.streamName()); assertEquals(SHARD_ID, request.shardId()); assertEquals(ShardIteratorType.AT_SEQUENCE_NUMBER.toString(), request.shardIteratorTypeAsString()); if (count == 2) { assertEquals(seqB, request.startingSequenceNumber()); } else { assertEquals(seqA, request.startingSequenceNumber()); } count++; } }
@Override public void advanceIteratorTo(final String sequenceNumber, final InitialPositionInStreamExtended initialPositionInStream) { if (sequenceNumber == null) { throw new IllegalArgumentException("SequenceNumber should not be null: shardId " + shardId); } final AWSExceptionManager exceptionManager = createExceptionManager(); GetShardIteratorRequest.Builder builder = KinesisRequestsBuilder.getShardIteratorRequestBuilder() .streamName(streamIdentifier.streamName()).shardId(shardId); GetShardIteratorRequest request = IteratorBuilder.request(builder, sequenceNumber, initialPositionInStream) .build(); final MetricsScope metricsScope = MetricsUtil.createMetricsWithOperation(metricsFactory, OPERATION); MetricsUtil.addStreamId(metricsScope, streamIdentifier); MetricsUtil.addShardId(metricsScope, shardId); boolean success = false; long startTime = System.currentTimeMillis(); try { try { nextIterator = getNextIterator(request); success = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (TimeoutException e) { throw new RetryableRetrievalException(e.getMessage(), e); } } catch (ResourceNotFoundException e) { log.info("Caught ResourceNotFoundException when getting an iterator for shard {}", streamAndShardId, e); nextIterator = null; } finally { MetricsUtil.addSuccessAndLatency(metricsScope, String.format("%s.%s", METRICS_PREFIX, "getShardIterator"), success, startTime, MetricsLevel.DETAILED); MetricsUtil.endScope(metricsScope); } if (nextIterator == null) { isShardEndReached = true; } this.lastKnownSequenceNumber = sequenceNumber; this.initialPositionInStream = initialPositionInStream; }
KinesisDataFetcher implements DataFetcher { @Override public void advanceIteratorTo(final String sequenceNumber, final InitialPositionInStreamExtended initialPositionInStream) { if (sequenceNumber == null) { throw new IllegalArgumentException("SequenceNumber should not be null: shardId " + shardId); } final AWSExceptionManager exceptionManager = createExceptionManager(); GetShardIteratorRequest.Builder builder = KinesisRequestsBuilder.getShardIteratorRequestBuilder() .streamName(streamIdentifier.streamName()).shardId(shardId); GetShardIteratorRequest request = IteratorBuilder.request(builder, sequenceNumber, initialPositionInStream) .build(); final MetricsScope metricsScope = MetricsUtil.createMetricsWithOperation(metricsFactory, OPERATION); MetricsUtil.addStreamId(metricsScope, streamIdentifier); MetricsUtil.addShardId(metricsScope, shardId); boolean success = false; long startTime = System.currentTimeMillis(); try { try { nextIterator = getNextIterator(request); success = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (TimeoutException e) { throw new RetryableRetrievalException(e.getMessage(), e); } } catch (ResourceNotFoundException e) { log.info("Caught ResourceNotFoundException when getting an iterator for shard {}", streamAndShardId, e); nextIterator = null; } finally { MetricsUtil.addSuccessAndLatency(metricsScope, String.format("%s.%s", METRICS_PREFIX, "getShardIterator"), success, startTime, MetricsLevel.DETAILED); MetricsUtil.endScope(metricsScope); } if (nextIterator == null) { isShardEndReached = true; } this.lastKnownSequenceNumber = sequenceNumber; this.initialPositionInStream = initialPositionInStream; } }
KinesisDataFetcher implements DataFetcher { @Override public void advanceIteratorTo(final String sequenceNumber, final InitialPositionInStreamExtended initialPositionInStream) { if (sequenceNumber == null) { throw new IllegalArgumentException("SequenceNumber should not be null: shardId " + shardId); } final AWSExceptionManager exceptionManager = createExceptionManager(); GetShardIteratorRequest.Builder builder = KinesisRequestsBuilder.getShardIteratorRequestBuilder() .streamName(streamIdentifier.streamName()).shardId(shardId); GetShardIteratorRequest request = IteratorBuilder.request(builder, sequenceNumber, initialPositionInStream) .build(); final MetricsScope metricsScope = MetricsUtil.createMetricsWithOperation(metricsFactory, OPERATION); MetricsUtil.addStreamId(metricsScope, streamIdentifier); MetricsUtil.addShardId(metricsScope, shardId); boolean success = false; long startTime = System.currentTimeMillis(); try { try { nextIterator = getNextIterator(request); success = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (TimeoutException e) { throw new RetryableRetrievalException(e.getMessage(), e); } } catch (ResourceNotFoundException e) { log.info("Caught ResourceNotFoundException when getting an iterator for shard {}", streamAndShardId, e); nextIterator = null; } finally { MetricsUtil.addSuccessAndLatency(metricsScope, String.format("%s.%s", METRICS_PREFIX, "getShardIterator"), success, startTime, MetricsLevel.DETAILED); MetricsUtil.endScope(metricsScope); } if (nextIterator == null) { isShardEndReached = true; } this.lastKnownSequenceNumber = sequenceNumber; this.initialPositionInStream = initialPositionInStream; } @Deprecated  KinesisDataFetcher(KinesisAsyncClient kinesisClient, String streamName, String shardId, int maxRecords, MetricsFactory metricsFactory);  KinesisDataFetcher(KinesisAsyncClient kinesisClient, DataFetcherProviderConfig kinesisDataFetcherProviderConfig); }
KinesisDataFetcher implements DataFetcher { @Override public void advanceIteratorTo(final String sequenceNumber, final InitialPositionInStreamExtended initialPositionInStream) { if (sequenceNumber == null) { throw new IllegalArgumentException("SequenceNumber should not be null: shardId " + shardId); } final AWSExceptionManager exceptionManager = createExceptionManager(); GetShardIteratorRequest.Builder builder = KinesisRequestsBuilder.getShardIteratorRequestBuilder() .streamName(streamIdentifier.streamName()).shardId(shardId); GetShardIteratorRequest request = IteratorBuilder.request(builder, sequenceNumber, initialPositionInStream) .build(); final MetricsScope metricsScope = MetricsUtil.createMetricsWithOperation(metricsFactory, OPERATION); MetricsUtil.addStreamId(metricsScope, streamIdentifier); MetricsUtil.addShardId(metricsScope, shardId); boolean success = false; long startTime = System.currentTimeMillis(); try { try { nextIterator = getNextIterator(request); success = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (TimeoutException e) { throw new RetryableRetrievalException(e.getMessage(), e); } } catch (ResourceNotFoundException e) { log.info("Caught ResourceNotFoundException when getting an iterator for shard {}", streamAndShardId, e); nextIterator = null; } finally { MetricsUtil.addSuccessAndLatency(metricsScope, String.format("%s.%s", METRICS_PREFIX, "getShardIterator"), success, startTime, MetricsLevel.DETAILED); MetricsUtil.endScope(metricsScope); } if (nextIterator == null) { isShardEndReached = true; } this.lastKnownSequenceNumber = sequenceNumber; this.initialPositionInStream = initialPositionInStream; } @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 advanceIteratorTo(final String sequenceNumber, final InitialPositionInStreamExtended initialPositionInStream) { if (sequenceNumber == null) { throw new IllegalArgumentException("SequenceNumber should not be null: shardId " + shardId); } final AWSExceptionManager exceptionManager = createExceptionManager(); GetShardIteratorRequest.Builder builder = KinesisRequestsBuilder.getShardIteratorRequestBuilder() .streamName(streamIdentifier.streamName()).shardId(shardId); GetShardIteratorRequest request = IteratorBuilder.request(builder, sequenceNumber, initialPositionInStream) .build(); final MetricsScope metricsScope = MetricsUtil.createMetricsWithOperation(metricsFactory, OPERATION); MetricsUtil.addStreamId(metricsScope, streamIdentifier); MetricsUtil.addShardId(metricsScope, shardId); boolean success = false; long startTime = System.currentTimeMillis(); try { try { nextIterator = getNextIterator(request); success = true; } catch (ExecutionException e) { throw exceptionManager.apply(e.getCause()); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (TimeoutException e) { throw new RetryableRetrievalException(e.getMessage(), e); } } catch (ResourceNotFoundException e) { log.info("Caught ResourceNotFoundException when getting an iterator for shard {}", streamAndShardId, e); nextIterator = null; } finally { MetricsUtil.addSuccessAndLatency(metricsScope, String.format("%s.%s", METRICS_PREFIX, "getShardIterator"), success, startTime, MetricsLevel.DETAILED); MetricsUtil.endScope(metricsScope); } if (nextIterator == null) { isShardEndReached = true; } this.lastKnownSequenceNumber = sequenceNumber; this.initialPositionInStream = initialPositionInStream; } @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);  }