Hi everyone, For the past weeks, we’ve been struggling with Kinesis ingestion using the Flink Kinesis connector, but the seemingly complete lack of similar reports makes us wonder if perhaps we misconfigured or mis-used the connector.
We’re using the connector to subscribe to streams varying from 1 to a 100 shards, and used the kinesis-scaling-utils to dynamically scale the Kinesis stream up and down during peak times. What we’ve noticed is that, while we were having closed shards, any Flink job restart with check- or save-point would result in shards being re-read from the event horizon, duplicating our events. We started checking the checkpoint state, and found that the shards were stored correctly with the proper sequence number (including for closed shards), but that upon restarts, the older closed shards would be read from the event horizon, as if their restored state would be ignored. In the end, we believe that we found the problem: in the FlinkKinesisConsumer’s run() method, we’re trying to find the shard returned from the KinesisDataFetcher against the shards’ metadata from the restoration point, but we do this via a containsKey() call, which means we’ll use the StreamShardMetadata’s equals() method. However, this checks for all properties, including the endingSequenceNumber, which might have changed between the restored state’s checkpoint and our data fetch, thus failing the equality check, failing the containsKey() check, and resulting in the shard being re-read from the event horizon, even though it was present in the restored state. We’ve created a workaround where we only check for the shardId and stream name to restore the state of the shards we’ve already seen, and this seems to work correctly. However, as pointed out above, the lack of similar reports makes us worried that we’ve misunderstood something, so we’d appreciate any feedback whether or not our report makes sense before we file a bug in the issue tracker. Much appreciated, -Phil -- "We cannot change the cards we are dealt, just how we play the hand." - Randy Pausch
