[kv] Support kv snapshot lease

[swuferhong]

Purpose

Linked issue: close #2171

Support kv snapshot lease

Brief change log

Tests

API and Format

Documentation

[Copilot]

Pull request overview

This PR adds support for KV snapshot consumers to prevent premature deletion of snapshots that are still being consumed. The implementation introduces a consumer registration mechanism where clients can register their interest in specific snapshots, preventing them from being deleted until consumption is complete or the consumer expires.

Key Changes:

• Introduced KvSnapshotConsumer data structure to track snapshot consumption per consumer
• Implemented KvSnapshotConsumerManager in the coordinator to manage consumer lifecycle and expiration
• Modified snapshot retention logic to be consumer-aware rather than based on a fixed retention count
• Added Flink source integration to automatically register/unregister consumers during checkpoint lifecycle

Reviewed changes

Copilot reviewed 65 out of 65 changed files in this pull request and generated 11 comments.

Show a summary per file

File	Description
KvSnapshotConsumer.java	New data structure tracking consumed snapshots per table/partition/bucket
KvSnapshotConsumerManager.java	Manager for consumer registration, unregistration, and expiration checking
CompletedSnapshotStore.java	Modified to use consumer-aware retention instead of fixed count
FlinkSourceEnumerator.java	Registers consumers when initializing splits, unregisters on checkpoint complete
FlinkSourceReader.java	Reports finished snapshot consumption via source events
ConfigOptions.java	Added configuration for consumer expiration checking and default expiration time
ZooKeeperClient.java	Added methods for persisting consumer state in ZooKeeper
Various test files	Removed fixed snapshot retention configuration, added consumer-based tests

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.