Github user glaksh100 commented on a diff in the pull request:
https://github.com/apache/flink/pull/6408#discussion_r204942574
--- Diff:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java
---
@@ -233,26 +225,68 @@ public void run() {
subscribedShard.getShard().getHashKeyRange().getEndingHashKey());
long recordBatchSizeBytes = 0L;
- long averageRecordSizeBytes = 0L;
-
for (UserRecord record :
fetchedRecords) {
recordBatchSizeBytes +=
record.getData().remaining();
deserializeRecordForCollectionAndUpdateState(record);
}
- if (useAdaptiveReads &&
!fetchedRecords.isEmpty()) {
- averageRecordSizeBytes =
recordBatchSizeBytes / fetchedRecords.size();
- maxNumberOfRecordsPerFetch =
getAdaptiveMaxRecordsPerFetch(averageRecordSizeBytes);
- }
-
nextShardItr =
getRecordsResult.getNextShardIterator();
+
+ long processingEndTimeNanos =
System.nanoTime();
+
+ long adjustmentEndTimeNanos =
adjustRunLoopFrequency(processingStartTimeNanos, processingEndTimeNanos);
+ long runLoopTimeNanos =
adjustmentEndTimeNanos - processingStartTimeNanos;
+ adaptRecordsToRead(runLoopTimeNanos,
fetchedRecords.size(), recordBatchSizeBytes);
+ processingStartTimeNanos =
adjustmentEndTimeNanos; // for next time through the loop
}
}
} catch (Throwable t) {
fetcherRef.stopWithError(t);
}
}
+ /**
+ * Adjusts loop timing to match target frequency if specified.
+ * @param processingStartTimeNanos The start time of the run loop "work"
+ * @param processingEndTimeNanos The end time of the run loop "work"
+ * @return The System.nanoTime() after the sleep (if any)
+ * @throws InterruptedException
+ */
+ protected long adjustRunLoopFrequency(long processingStartTimeNanos,
long processingEndTimeNanos)
+ throws InterruptedException {
+ long endTimeNanos = processingEndTimeNanos;
+ if (fetchIntervalMillis != 0) {
+ long processingTimeNanos = processingEndTimeNanos -
processingStartTimeNanos;
+ long sleepTimeMillis = fetchIntervalMillis -
(processingTimeNanos / 1_000_000);
+ if (sleepTimeMillis > 0) {
+ Thread.sleep(sleepTimeMillis);
+ endTimeNanos = System.nanoTime();
+ }
+ }
+ return endTimeNanos;
+ }
+
+ /**
+ * Calculates how many records to read each time through the loop based
on a target throughput
+ * and the measured frequenecy of the loop.
+ * @param runLoopTimeNanos The total time of one pass through the loop
+ * @param numRecords The number of records of the last read operation
+ * @param recordBatchSizeBytes The total batch size of the last read
operation
+ */
+ protected int adaptRecordsToRead(long runLoopTimeNanos, int numRecords,
long recordBatchSizeBytes) {
+ 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 not more than 10000L
+ maxNumberOfRecordsPerFetch =
Math.min(maxNumberOfRecordsPerFetch,
ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_MAX);
+ }
+ return maxNumberOfRecordsPerFetch;
--- End diff --
Oops, thanks for catching. Updated to use the return value and also to use
a local variable in the method to avoid re-assigning the class variable
`maxNumberOfRecordsPerFetch` twice.
---
