[FLINK-8542] Do not indefinitely store closed shard's state in the FlinkKinesisConsumer

• Type: Improvement
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Component/s: Connectors / Kinesis
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Tzu-Li (Gordon) Tai
• Created: 01/Feb/18 10:58
• Updated: 09/Dec/21 22:38

Description

See original discussion here: https://github.com/apache/flink/pull/5337

Currently, the Kinesis consumer keeps a list of (StreamShardMetadata, SequenceNumber) as its state. That list also contains all shards that have been closed already, and is kept in the state indefinitely so that on restore, we know that a closed shard is already fully consumed,

The downside of this, is that the state size of the Kinesis consumer can basically grow without bounds, as the consumed Kinesis streams are resharded and more and more closed shards are present.

Some possible solutions have been discussed in the linked PR comments.