bube commented on code in PR #254:
URL:
https://github.com/apache/flink-connector-aws/pull/254#discussion_r3842900477
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/KinesisShardSplitReaderBase.java:
##########
@@ -179,31 +184,47 @@ private boolean
skipUntilScheduledFetchTime(KinesisShardSplitState splitState)
}
/**
- * Schedules next fetch time, to be called immediately on the result of a
fetchRecords() call.
+ * Schedules the next fetch time. To be called immediately after a
fetchRecords() call.
*
- * <p>If recordBatch does not contain records, next fetchRecords() is
scheduled. Before
- * scheduled time, fetcher thread will skip fetching (and have small
sleep) for the split.
+ * <p>If recordBatch is null or contains no records, the next fetch is
scheduled using
+ * emptyRecordsIntervalMillis. Before the scheduled time, the fetcher
thread will skip fetching
+ * (and have a small sleep) for the split.
*
- * <p>If recordBatch is not empty, next fetchRecords() time is not
scheduled resulting in next
- * fetch on the split is performed at first opportunity.
+ * <p>If recordBatch is not empty, the next fetch is scheduled using
+ * nonEmptyRecordsIntervalMillis. When that interval is zero (the
default), no fetch time is
+ * scheduled and the next fetch on the split is performed at the first
opportunity.
Review Comment:
Confirmed. The "next opportunity" path is outside these code changes.
`fetch()` calls `skipUntilScheduledFetchTime` to check if should skip.
`skipUntilScheduledFetchTime()` checks if a next fetch time has been scheduled
by `scheduleNextFetchTime()`:
```
if (scheduledFetchTimes.containsKey(splitState)
&& scheduledFetchTimes.get(splitState) >
System.currentTimeMillis())
```
Scheduling a 0 (i.e. NOW + 0) would be redundant (and perform slightly
worse) because `scheduledFetchTimes.get(splitState) >
System.currentTimeMillis()` always would be false. Not setting it allows the
method to return quickly already at
`scheduledFetchTimes.containsKey(splitState)` returning `false`.
I pinned the existing behaviour with the test
`KinesisShardSplitReaderBaseTest#testNoGetRecordsIntervalForNonEmptySourceByDefault`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
