@Test @Ignore public void testFullCacheByteSize() { record = Record.builder().data(createByteBufferWithSize(SIZE_1_MB)).build(); when(records.size()).thenReturn(500); records.add(record); getRecordsCache.start(sequenceNumber, initialPosition); sleep(2000); verify(getRecordsRetrievalStrategy, times(3)).getRecords(eq(MAX_RECORDS_PER_CALL)); assertEquals(spyQueue.size(), 3); }
@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);  }