@Test @Ignore public void testRestartIterator() throws Exception { GetRecordsResponse getRecordsResult = mock(GetRecordsResponse.class); GetRecordsResponse restartGetRecordsResponse = makeGetRecordsResponse(null, null).get(anyLong(), any(TimeUnit.class)); Record record = mock(Record.class); final String nextShardIterator = "NextShardIterator"; final String sequenceNumber = "SequenceNumber"; when(getRecordsResult.records()).thenReturn(Collections.singletonList(record)); when(getRecordsResult.nextShardIterator()).thenReturn(nextShardIterator); when(record.sequenceNumber()).thenReturn(sequenceNumber); kinesisDataFetcher.initialize(InitialPositionInStream.LATEST.toString(), INITIAL_POSITION_LATEST); assertEquals(getRecordsResult, kinesisDataFetcher.getRecords().accept()); kinesisDataFetcher.restartIterator(); assertEquals(restartGetRecordsResponse, kinesisDataFetcher.getRecords().accept()); }
@Override public void restartIterator() { if (StringUtils.isEmpty(lastKnownSequenceNumber) || initialPositionInStream == null) { throw new IllegalStateException( "Make sure to initialize the KinesisDataFetcher before restarting the iterator."); } advanceIteratorTo(lastKnownSequenceNumber, initialPositionInStream); }
KinesisDataFetcher implements DataFetcher { @Override public void restartIterator() { if (StringUtils.isEmpty(lastKnownSequenceNumber) || initialPositionInStream == null) { throw new IllegalStateException( "Make sure to initialize the KinesisDataFetcher before restarting the iterator."); } advanceIteratorTo(lastKnownSequenceNumber, initialPositionInStream); } }
KinesisDataFetcher implements DataFetcher { @Override public void restartIterator() { if (StringUtils.isEmpty(lastKnownSequenceNumber) || initialPositionInStream == null) { throw new IllegalStateException( "Make sure to initialize the KinesisDataFetcher before restarting the iterator."); } advanceIteratorTo(lastKnownSequenceNumber, initialPositionInStream); } @Deprecated  KinesisDataFetcher(KinesisAsyncClient kinesisClient, String streamName, String shardId, int maxRecords, MetricsFactory metricsFactory);  KinesisDataFetcher(KinesisAsyncClient kinesisClient, DataFetcherProviderConfig kinesisDataFetcherProviderConfig); }
KinesisDataFetcher implements DataFetcher { @Override public void restartIterator() { if (StringUtils.isEmpty(lastKnownSequenceNumber) || initialPositionInStream == null) { throw new IllegalStateException( "Make sure to initialize the KinesisDataFetcher before restarting the iterator."); } advanceIteratorTo(lastKnownSequenceNumber, 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 restartIterator() { if (StringUtils.isEmpty(lastKnownSequenceNumber) || initialPositionInStream == null) { throw new IllegalStateException( "Make sure to initialize the KinesisDataFetcher before restarting the iterator."); } advanceIteratorTo(lastKnownSequenceNumber, 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);  }