[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.

[platinumhamburg]

@swuferhong Thank you for undertaking this complex work, I've left some review comments.

[platinumhamburg]

I think it should be implemented at the table level so we can quickly identify the affected business when lease leaks occur.

[swuferhong]

Sure, I was thinking the same thing. I'll add a TODO for now, and this will be implemented in the next PR since this PR is already too large. Trace by: #2297

[platinumhamburg]

Ditto—“drop” isn’t the most appropriate verb to use with “lease.”

[swuferhong]

Do you intend to rename this API to release, or just update the comment? According to the FIP, this API is indeed named dropKvSnapshotLease, and its behavior genuinely performs a "drop" operation.

[platinumhamburg]

I noticed we're using Jackson's very low-level APIs, which makes the serialization/deserialization logic quite complex. What's the reason behind this approach?

[swuferhong]

This implementation is indeed quite old—it was originally based on Flink's serialization. We could consider refactoring it, but perhaps not in this PR?