bube commented on code in PR #254:
URL:
https://github.com/apache/flink-connector-aws/pull/254#discussion_r3842682090
##########
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.
*
- * @param splitState splitState on which the fetchRecords() was called on
- * @param recordBatch recordBatch returned by fetchRecords()
+ * @param splitState split state the fetchRecords() call was made for
+ * @param recordBatch record batch returned by fetchRecords()
*/
private void scheduleNextFetchTime(KinesisShardSplitState splitState,
RecordBatch recordBatch) {
if (recordBatch == null || recordBatch.getRecords().isEmpty()) {
+ long scheduledGetRecordTimeMillis = scheduleAt(splitState,
emptyRecordsIntervalMillis);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Fetched zero records from split {}, scheduling next
fetch at {}",
+ splitState.getSplitId(),
+ new Date(scheduledGetRecordTimeMillis).toInstant());
Review Comment:
`Instant.toString()` uses ISO-8601 (e.g. 2011-12-03T10:15:30Z). However, I
had blindly copied the log message format from the original code which uses the
ancient `java.util.Date` class. Instead, I'll use `Instant.ofEpochMilli()`
directly. No need to instantiate a `java.util.Date` object for this.
--
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]
