@Test public void testExpiredIteratorException() { log.info("Starting tests"); when(getRecordsRetrievalStrategy.getRecords(MAX_RECORDS_PER_CALL)).thenThrow(ExpiredIteratorException.class) .thenReturn(getRecordsResponse); getRecordsCache.start(sequenceNumber, initialPosition); doNothing().when(dataFetcher).restartIterator(); blockUntilRecordsAvailable(() -> evictPublishedEvent(getRecordsCache, "shardId"), 1000L); sleep(1000); verify(dataFetcher).restartIterator(); }
@Override public void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended) { if (executorService.isShutdown()) { throw new IllegalStateException("ExecutorService has been shutdown."); } if (!started) { log.info("{} : Starting Prefetching thread and initializing publisher session.", streamAndShardId); publisherSession.init(extendedSequenceNumber, initialPositionInStreamExtended); executorService.execute(defaultGetRecordsCacheDaemon); } else { log.info("{} : Skipping publisher start as it was already started.", streamAndShardId); } started = true; }
PrefetchRecordsPublisher implements RecordsPublisher { @Override public void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended) { if (executorService.isShutdown()) { throw new IllegalStateException("ExecutorService has been shutdown."); } if (!started) { log.info("{} : Starting Prefetching thread and initializing publisher session.", streamAndShardId); publisherSession.init(extendedSequenceNumber, initialPositionInStreamExtended); executorService.execute(defaultGetRecordsCacheDaemon); } else { log.info("{} : Skipping publisher start as it was already started.", streamAndShardId); } started = true; } }
PrefetchRecordsPublisher implements RecordsPublisher { @Override public void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended) { if (executorService.isShutdown()) { throw new IllegalStateException("ExecutorService has been shutdown."); } if (!started) { log.info("{} : Starting Prefetching thread and initializing publisher session.", streamAndShardId); publisherSession.init(extendedSequenceNumber, initialPositionInStreamExtended); executorService.execute(defaultGetRecordsCacheDaemon); } else { log.info("{} : Skipping publisher start as it was already started.", streamAndShardId); } started = true; } PrefetchRecordsPublisher(final int maxPendingProcessRecordsInput, final int maxByteSize, final int maxRecordsCount,
                                    final int maxRecordsPerCall,
                                    @NonNull final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy,
                                    @NonNull final ExecutorService executorService,
                                    final long idleMillisBetweenCalls,
                                    @NonNull final MetricsFactory metricsFactory,
                                    @NonNull final String operation,
                                    @NonNull final String shardId); }
PrefetchRecordsPublisher implements RecordsPublisher { @Override public void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended) { if (executorService.isShutdown()) { throw new IllegalStateException("ExecutorService has been shutdown."); } if (!started) { log.info("{} : Starting Prefetching thread and initializing publisher session.", streamAndShardId); publisherSession.init(extendedSequenceNumber, initialPositionInStreamExtended); executorService.execute(defaultGetRecordsCacheDaemon); } else { log.info("{} : Skipping publisher start as it was already started.", streamAndShardId); } started = true; } PrefetchRecordsPublisher(final int maxPendingProcessRecordsInput, final int maxByteSize, final int maxRecordsCount,
                                    final int maxRecordsPerCall,
                                    @NonNull final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy,
                                    @NonNull final ExecutorService executorService,
                                    final long idleMillisBetweenCalls,
                                    @NonNull final MetricsFactory metricsFactory,
                                    @NonNull final String operation,
                                    @NonNull final String shardId); @Override void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended); @Override void shutdown(); @Override RequestDetails getLastSuccessfulRequestDetails(); @Override void restartFrom(RecordsRetrieved recordsRetrieved); @Override void subscribe(Subscriber<? super RecordsRetrieved> s); @Override synchronized void notify(RecordsDeliveryAck recordsDeliveryAck); }
PrefetchRecordsPublisher implements RecordsPublisher { @Override public void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended) { if (executorService.isShutdown()) { throw new IllegalStateException("ExecutorService has been shutdown."); } if (!started) { log.info("{} : Starting Prefetching thread and initializing publisher session.", streamAndShardId); publisherSession.init(extendedSequenceNumber, initialPositionInStreamExtended); executorService.execute(defaultGetRecordsCacheDaemon); } else { log.info("{} : Skipping publisher start as it was already started.", streamAndShardId); } started = true; } PrefetchRecordsPublisher(final int maxPendingProcessRecordsInput, final int maxByteSize, final int maxRecordsCount,
                                    final int maxRecordsPerCall,
                                    @NonNull final GetRecordsRetrievalStrategy getRecordsRetrievalStrategy,
                                    @NonNull final ExecutorService executorService,
                                    final long idleMillisBetweenCalls,
                                    @NonNull final MetricsFactory metricsFactory,
                                    @NonNull final String operation,
                                    @NonNull final String shardId); @Override void start(ExtendedSequenceNumber extendedSequenceNumber, InitialPositionInStreamExtended initialPositionInStreamExtended); @Override void shutdown(); @Override RequestDetails getLastSuccessfulRequestDetails(); @Override void restartFrom(RecordsRetrieved recordsRetrieved); @Override void subscribe(Subscriber<? super RecordsRetrieved> s); @Override synchronized void notify(RecordsDeliveryAck recordsDeliveryAck);  }