private int adaptRecordsToRead ( long runLoopTimeNanos , int numRecords , long recordBatchSizeBytes , int maxNumberOfRecordsPerFetch ) { if ( useAdaptiveReads && numRecords != 0 && runLoopTimeNanos != 0 ) { long averageRecordSizeBytes = recordBatchSizeBytes / numRecords ; // Adjust number of records to fetch from the shard depending on current average record size // to optimize 2 Mb / sec read limits double loopFrequencyHz = 1000000000.0d / runLoopTimeNanos ; double bytesPerRead = KINESIS_SHARD_BYTES_PER_SECOND_LIMIT / loopFrequencyHz ; maxNumberOfRecordsPerFetch = ( int ) ( bytesPerRead / averageRecordSizeBytes ) ; // Ensure the value is greater than 0 and not more than 10000L maxNumberOfRecordsPerFetch = Math . max ( 1 , Math . min ( maxNumberOfRecordsPerFetch , ConsumerConfigConstants . DEFAULT_SHARD_GETRECORDS_MAX ) ) ; // Set metrics shardMetricsReporter . setAverageRecordSizeBytes ( averageRecordSizeBytes ) ; shardMetricsReporter . setLoopFrequencyHz ( loopFrequencyHz ) ; shardMetricsReporter . setBytesPerRead ( bytesPerRead ) ; } return maxNumberOfRecordsPerFetch ; }
Calculates how many records to read each time through the loop based on a target throughput and the measured frequenecy of the loop .