@Test public void testGetRecordsWithShardEnd() { records = new ArrayList<>(); final List<KinesisClientRecord> expectedRecords = new ArrayList<>(); List<ChildShard> childShards = new ArrayList<>(); List<String> parentShards = new ArrayList<>(); parentShards.add("shardId"); ChildShard leftChild = ChildShard.builder() .shardId("shardId-000000000001") .parentShards(parentShards) .hashKeyRange(ShardObjectHelper.newHashKeyRange("0", "49")) .build(); ChildShard rightChild = ChildShard.builder() .shardId("shardId-000000000002") .parentShards(parentShards) .hashKeyRange(ShardObjectHelper.newHashKeyRange("50", "99")) .build(); childShards.add(leftChild); childShards.add(rightChild); GetRecordsResponse response = GetRecordsResponse.builder().records(records).childShards(childShards).build(); when(getRecordsRetrievalStrategy.getRecords(eq(MAX_RECORDS_PER_CALL))).thenReturn(response); when(dataFetcher.isShardEndReached()).thenReturn(true); getRecordsCache.start(sequenceNumber, initialPosition); ProcessRecordsInput result = blockUntilRecordsAvailable(() -> evictPublishedEvent(getRecordsCache, "shardId"), 1000L) .processRecordsInput(); assertEquals(expectedRecords, result.records()); assertEquals(childShards, result.childShards()); assertTrue(result.isAtShardEnd()); }
@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);  }