gguptp commented on code in PR #195:
URL:
https://github.com/apache/flink-connector-aws/pull/195#discussion_r2028304276
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/reader/polling/PollingKinesisShardSplitReader.java:
##########
@@ -40,6 +47,14 @@ public class PollingKinesisShardSplitReader extends
KinesisShardSplitReaderBase
private final Configuration configuration;
private final int maxRecordsToGet;
+ private final long getRecordsIntervalMillis;
+
+ private final long idleSourceGetRecordsIntervalMillis;
+
+ private long scheduledGetRecordTimeMillis = 0;
Review Comment:
hmm, given `PollingKinesisShardSplitReader` is a child class of
`KinesisShardSplitReaderBase` doesnt this variable get shared between all the
shards rather than just one particular shard?
Won't the behaviour become like this:
Shard 1 - empty poll (scheduledGetRecordTimeMillis = T+250 millis)
Shard 2 - empty poll (scheduledGetRecordTimeMillis = T1+250 millis)
...
and so on
now theres a chance that shard 1 comes up later and although it should have
been called with getrecords but since n shards came and the new
scheduledGetRecordTimeMillis becomes much more than the actual
shceduledGetRecordTimeMillis for shard 1 and this might lead to starvation (no
getrecords calls at all?) of all the shards?
We should probably write a UT for this case as well i believe
--
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]
