tzulitai commented on a change in pull request #7601: [FLINK-11164] Check for
sentinel values when creating new Kinesis ShardIterator
URL: https://github.com/apache/flink/pull/7601#discussion_r252220868
##########
File path:
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java
##########
@@ -380,7 +381,19 @@ private GetRecordsResult getRecords(String shardItr, int
maxNumberOfRecords) thr
} catch (ExpiredIteratorException eiEx) {
LOG.warn("Encountered an unexpected expired
iterator {} for shard {};" +
" refreshing the iterator ...",
shardItr, subscribedShard);
- shardItr =
kinesis.getShardIterator(subscribedShard,
ShardIteratorType.AFTER_SEQUENCE_NUMBER.toString(),
lastSequenceNum.getSequenceNumber());
+ if
(SentinelSequenceNumber.getAllSentinelSequenceNumbers().contains(lastSequenceNum))
{
+ // we cannot create a shard iterator
because we still have an initial sentinel
+ // value as the current sequence number
+ LOG.warn(
Review comment:
This doesn't fix the problem.
If we only log a warning in this case, and the shard iterator remains to be
an expired one, then this will basically just fall into an infinite loop
(because the expired iterator will still be attempted to be used).
Therefore, we still need to update the iterator.
What we need to do here is that instead of calling
`kinesis.getShardIterator(...)` directly with
`lastSequenceNum.getSequenceNumber()` to update the shard iterator, we do
something similar to what is done here:
https://github.com/apache/flink/blob/1d3fe964b4eec520c14dcb9e810a4e84bedb0110/flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/ShardConsumer.java#L147
Shortly put, if the sequence number is still a sentinel placeholder value,
depending on the value, calls to `kinesis.getShardIterator` would be a bit
different.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
