@Test public void testCorrectNumOfCollectedRecordsAndUpdatedStateWithAdaptiveReads() { Properties consumerProperties = new Properties(); consumerProperties.put("flink.shard.adaptivereads", "true"); StreamShardHandle fakeToBeConsumedShard = getMockStreamShard("fakeStream", 0); LinkedList<KinesisStreamShardState> subscribedShardsStateUnderTest = new LinkedList<>(); subscribedShardsStateUnderTest.add( new KinesisStreamShardState(KinesisDataFetcher.convertToStreamShardMetadata(fakeToBeConsumedShard), fakeToBeConsumedShard, new SequenceNumber("fakeStartingState"))); TestSourceContext<String> sourceContext = new TestSourceContext<>(); TestableKinesisDataFetcher<String> fetcher = new TestableKinesisDataFetcher<>( Collections.singletonList("fakeStream"), sourceContext, consumerProperties, new KinesisDeserializationSchemaWrapper<>(new SimpleStringSchema()), 10, 2, new AtomicReference<>(), subscribedShardsStateUnderTest, KinesisDataFetcher.createInitialSubscribedStreamsToLastDiscoveredShardsState(Collections.singletonList("fakeStream")), Mockito.mock(KinesisProxyInterface.class)); int shardIndex = fetcher.registerNewSubscribedShardState(subscribedShardsStateUnderTest.get(0)); new ShardConsumer<>( fetcher, shardIndex, subscribedShardsStateUnderTest.get(0).getStreamShardHandle(), subscribedShardsStateUnderTest.get(0).getLastProcessedSequenceNum(), FakeKinesisBehavioursFactory.initialNumOfRecordsAfterNumOfGetRecordsCallsWithAdaptiveReads(10, 2, 500L), new ShardMetricsReporter()).run(); assertEquals(50, sourceContext.getCollectedOutputs().size()); assertEquals( SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get(), subscribedShardsStateUnderTest.get(0).getLastProcessedSequenceNum()); }
@SuppressWarnings("unchecked") @Override public void run() { try { String nextShardItr = getShardIterator(lastSequenceNum); long processingStartTimeNanos = System.nanoTime(); while (isRunning()) { if (nextShardItr == null) { fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()); break; } else { shardMetricsReporter.setMaxNumberOfRecordsPerFetch(maxNumberOfRecordsPerFetch); GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch); List<Record> aggregatedRecords = getRecordsResult.getRecords(); int numberOfAggregatedRecords = aggregatedRecords.size(); shardMetricsReporter.setNumberOfAggregatedRecords(numberOfAggregatedRecords); List<UserRecord> fetchedRecords = deaggregateRecords( aggregatedRecords, subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); long recordBatchSizeBytes = 0L; for (UserRecord record : fetchedRecords) { recordBatchSizeBytes += record.getData().remaining(); deserializeRecordForCollectionAndUpdateState(record); } int numberOfDeaggregatedRecords = fetchedRecords.size(); shardMetricsReporter.setNumberOfDeaggregatedRecords(numberOfDeaggregatedRecords); nextShardItr = getRecordsResult.getNextShardIterator(); long adjustmentEndTimeNanos = adjustRunLoopFrequency(processingStartTimeNanos, System.nanoTime()); long runLoopTimeNanos = adjustmentEndTimeNanos - processingStartTimeNanos; maxNumberOfRecordsPerFetch = adaptRecordsToRead(runLoopTimeNanos, fetchedRecords.size(), recordBatchSizeBytes, maxNumberOfRecordsPerFetch); shardMetricsReporter.setRunLoopTimeNanos(runLoopTimeNanos); processingStartTimeNanos = adjustmentEndTimeNanos; } } } catch (Throwable t) { fetcherRef.stopWithError(t); } }
ShardConsumer implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { try { String nextShardItr = getShardIterator(lastSequenceNum); long processingStartTimeNanos = System.nanoTime(); while (isRunning()) { if (nextShardItr == null) { fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()); break; } else { shardMetricsReporter.setMaxNumberOfRecordsPerFetch(maxNumberOfRecordsPerFetch); GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch); List<Record> aggregatedRecords = getRecordsResult.getRecords(); int numberOfAggregatedRecords = aggregatedRecords.size(); shardMetricsReporter.setNumberOfAggregatedRecords(numberOfAggregatedRecords); List<UserRecord> fetchedRecords = deaggregateRecords( aggregatedRecords, subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); long recordBatchSizeBytes = 0L; for (UserRecord record : fetchedRecords) { recordBatchSizeBytes += record.getData().remaining(); deserializeRecordForCollectionAndUpdateState(record); } int numberOfDeaggregatedRecords = fetchedRecords.size(); shardMetricsReporter.setNumberOfDeaggregatedRecords(numberOfDeaggregatedRecords); nextShardItr = getRecordsResult.getNextShardIterator(); long adjustmentEndTimeNanos = adjustRunLoopFrequency(processingStartTimeNanos, System.nanoTime()); long runLoopTimeNanos = adjustmentEndTimeNanos - processingStartTimeNanos; maxNumberOfRecordsPerFetch = adaptRecordsToRead(runLoopTimeNanos, fetchedRecords.size(), recordBatchSizeBytes, maxNumberOfRecordsPerFetch); shardMetricsReporter.setRunLoopTimeNanos(runLoopTimeNanos); processingStartTimeNanos = adjustmentEndTimeNanos; } } } catch (Throwable t) { fetcherRef.stopWithError(t); } } }
ShardConsumer implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { try { String nextShardItr = getShardIterator(lastSequenceNum); long processingStartTimeNanos = System.nanoTime(); while (isRunning()) { if (nextShardItr == null) { fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()); break; } else { shardMetricsReporter.setMaxNumberOfRecordsPerFetch(maxNumberOfRecordsPerFetch); GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch); List<Record> aggregatedRecords = getRecordsResult.getRecords(); int numberOfAggregatedRecords = aggregatedRecords.size(); shardMetricsReporter.setNumberOfAggregatedRecords(numberOfAggregatedRecords); List<UserRecord> fetchedRecords = deaggregateRecords( aggregatedRecords, subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); long recordBatchSizeBytes = 0L; for (UserRecord record : fetchedRecords) { recordBatchSizeBytes += record.getData().remaining(); deserializeRecordForCollectionAndUpdateState(record); } int numberOfDeaggregatedRecords = fetchedRecords.size(); shardMetricsReporter.setNumberOfDeaggregatedRecords(numberOfDeaggregatedRecords); nextShardItr = getRecordsResult.getNextShardIterator(); long adjustmentEndTimeNanos = adjustRunLoopFrequency(processingStartTimeNanos, System.nanoTime()); long runLoopTimeNanos = adjustmentEndTimeNanos - processingStartTimeNanos; maxNumberOfRecordsPerFetch = adaptRecordsToRead(runLoopTimeNanos, fetchedRecords.size(), recordBatchSizeBytes, maxNumberOfRecordsPerFetch); shardMetricsReporter.setRunLoopTimeNanos(runLoopTimeNanos); processingStartTimeNanos = adjustmentEndTimeNanos; } } } catch (Throwable t) { fetcherRef.stopWithError(t); } } ShardConsumer(KinesisDataFetcher<T> fetcherRef,
						Integer subscribedShardStateIndex,
						StreamShardHandle subscribedShard,
						SequenceNumber lastSequenceNum,
						KinesisProxyInterface kinesis,
						ShardMetricsReporter shardMetricsReporter); }
ShardConsumer implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { try { String nextShardItr = getShardIterator(lastSequenceNum); long processingStartTimeNanos = System.nanoTime(); while (isRunning()) { if (nextShardItr == null) { fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()); break; } else { shardMetricsReporter.setMaxNumberOfRecordsPerFetch(maxNumberOfRecordsPerFetch); GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch); List<Record> aggregatedRecords = getRecordsResult.getRecords(); int numberOfAggregatedRecords = aggregatedRecords.size(); shardMetricsReporter.setNumberOfAggregatedRecords(numberOfAggregatedRecords); List<UserRecord> fetchedRecords = deaggregateRecords( aggregatedRecords, subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); long recordBatchSizeBytes = 0L; for (UserRecord record : fetchedRecords) { recordBatchSizeBytes += record.getData().remaining(); deserializeRecordForCollectionAndUpdateState(record); } int numberOfDeaggregatedRecords = fetchedRecords.size(); shardMetricsReporter.setNumberOfDeaggregatedRecords(numberOfDeaggregatedRecords); nextShardItr = getRecordsResult.getNextShardIterator(); long adjustmentEndTimeNanos = adjustRunLoopFrequency(processingStartTimeNanos, System.nanoTime()); long runLoopTimeNanos = adjustmentEndTimeNanos - processingStartTimeNanos; maxNumberOfRecordsPerFetch = adaptRecordsToRead(runLoopTimeNanos, fetchedRecords.size(), recordBatchSizeBytes, maxNumberOfRecordsPerFetch); shardMetricsReporter.setRunLoopTimeNanos(runLoopTimeNanos); processingStartTimeNanos = adjustmentEndTimeNanos; } } } catch (Throwable t) { fetcherRef.stopWithError(t); } } ShardConsumer(KinesisDataFetcher<T> fetcherRef,
						Integer subscribedShardStateIndex,
						StreamShardHandle subscribedShard,
						SequenceNumber lastSequenceNum,
						KinesisProxyInterface kinesis,
						ShardMetricsReporter shardMetricsReporter); @SuppressWarnings("unchecked") @Override void run(); }
ShardConsumer implements Runnable { @SuppressWarnings("unchecked") @Override public void run() { try { String nextShardItr = getShardIterator(lastSequenceNum); long processingStartTimeNanos = System.nanoTime(); while (isRunning()) { if (nextShardItr == null) { fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get()); break; } else { shardMetricsReporter.setMaxNumberOfRecordsPerFetch(maxNumberOfRecordsPerFetch); GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch); List<Record> aggregatedRecords = getRecordsResult.getRecords(); int numberOfAggregatedRecords = aggregatedRecords.size(); shardMetricsReporter.setNumberOfAggregatedRecords(numberOfAggregatedRecords); List<UserRecord> fetchedRecords = deaggregateRecords( aggregatedRecords, subscribedShard.getShard().getHashKeyRange().getStartingHashKey(), subscribedShard.getShard().getHashKeyRange().getEndingHashKey()); long recordBatchSizeBytes = 0L; for (UserRecord record : fetchedRecords) { recordBatchSizeBytes += record.getData().remaining(); deserializeRecordForCollectionAndUpdateState(record); } int numberOfDeaggregatedRecords = fetchedRecords.size(); shardMetricsReporter.setNumberOfDeaggregatedRecords(numberOfDeaggregatedRecords); nextShardItr = getRecordsResult.getNextShardIterator(); long adjustmentEndTimeNanos = adjustRunLoopFrequency(processingStartTimeNanos, System.nanoTime()); long runLoopTimeNanos = adjustmentEndTimeNanos - processingStartTimeNanos; maxNumberOfRecordsPerFetch = adaptRecordsToRead(runLoopTimeNanos, fetchedRecords.size(), recordBatchSizeBytes, maxNumberOfRecordsPerFetch); shardMetricsReporter.setRunLoopTimeNanos(runLoopTimeNanos); processingStartTimeNanos = adjustmentEndTimeNanos; } } } catch (Throwable t) { fetcherRef.stopWithError(t); } } ShardConsumer(KinesisDataFetcher<T> fetcherRef,
						Integer subscribedShardStateIndex,
						StreamShardHandle subscribedShard,
						SequenceNumber lastSequenceNum,
						KinesisProxyInterface kinesis,
						ShardMetricsReporter shardMetricsReporter); @SuppressWarnings("unchecked") @Override void run();  }