[FLINK-9691] [kinesis] Modify runloop to try to track a particular ge…

…tRecords() frequency. This closes apache#6290
laubersder · Jul 10, 2018 · 7953a2e · 7953a2e
1 parent abfdc1a
commit 7953a2e
Showing 1 changed file with 7 additions and 1 deletion.
diff --git a/.../src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java b/.../src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java
@@ -199,6 +199,7 @@ public void run() {
  }
  }
 
+ long lastTimeNanos = 0;
  while (isRunning()) {
  if (nextShardItr == null) {
  fetcherRef.updateState(subscribedShardStateIndex, SentinelSequenceNumber.SENTINEL_SHARD_ENDING_SEQUENCE_NUM.get());
@@ -207,7 +208,12 @@ public void run() {
  break;
  } else {
  if (fetchIntervalMillis != 0) {
- Thread.sleep(fetchIntervalMillis);
+ long elapsedTimeNanos = System.nanoTime() - lastTimeNanos;
+ long sleepTimeMillis = fetchIntervalMillis - (elapsedTimeNanos / 1_000_000);
+ if (sleepTimeMillis > 0) {
+ Thread.sleep(sleepTimeMillis);
+ }
+ lastTimeNanos = System.nanoTime();
  }
 
  GetRecordsResult getRecordsResult = getRecords(nextShardItr, maxNumberOfRecordsPerFetch);