[FLINK-10154][connectors] Make sure we always read at least one recor…

…d in KinesisConnector. This closes #6564.
apache · Aug 16, 2018 · 02152e7 · 02152e7
1 parent 92815e7
commit 02152e7
Showing 1 changed file with 2 additions and 2 deletions.
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
@@ -278,8 +278,8 @@ private int adaptRecordsToRead(long runLoopTimeNanos, int numRecords, long recor
 			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);
+			// Ensure the value is greater than 0 and not more than 10000L
+			maxNumberOfRecordsPerFetch = Math.max(1, Math.min(maxNumberOfRecordsPerFetch, ConsumerConfigConstants.DEFAULT_SHARD_GETRECORDS_MAX));
 		}
 		return maxNumberOfRecordsPerFetch;
 	}