@PublicEvolving
public interface Admin extends AutoCloseable {      

	// --- the following is added method 


   /**
     * Acquires a lease for specific KV snapshots of the given tableBuckets asynchronously.
     * <p> Once acquired, the specified KV snapshots will be protected from garbage collection 
     * for the duration of the {@code leaseDuration}. The client must call 
     * {@link #renewKvSnapshotLease} periodically to keep the lease alive, or call 
     * {@link #releaseKvSnapshotLease} to release the lock early when reading is finished.
     * 
	 * <p>If the lease expires (no renew received within duration), the server is free to 
     * delete the snapshot files.
     *
     * <p>The following exceptions can be anticipated when calling {@code get()} on returned future:
     * <ul>
     * <li>{@link TableNotExistException} if the table does not exist.
     * <li>{@link PartitionNotExistException} if the partition does not exist.
     * </ul>
     *
     * @param leaseId   The unique ID for this lease session (usually a UUID generated by client).
     * @param snapshotIds The snapshots to consume, a map from TableBucket to kvSnapshotId.
     * @param leaseDuration  The duration (in milliseconds) for which the snapshots should be kept.
     * @return The result of the acquire operation, containing any buckets that failed to be locked.
     */
    CompletableFuture<AcquireKvSnapshotLeaseResult> acquireKvSnapshotLease(
		String leaseId, 
        Map<TableBucket, Long> snapshotIds, 
        long leaseDuration);

    /**
     * Renews the lease for the given leaseId asynchronously.
     * 
	 * <p> This acts as a heartbeat. It resets the expiration time for ALL tableBuckets currently 
     * associated with this {@code leaseId}. 
     *
     * <p>The following exceptions can be anticipated:
     * <ul>
     * <li>{@link KvSnapshotLeaseNotFoundException} if the leaseId does not exist or has already expired.
     * </ul>
     *
     * @param leaseId       The lease id to renew.
     * @param leaseDuration The new duration (in milliseconds) from now.
     */
    CompletableFuture<Void> renewKvSnapshotLease(
            String leaseId, 
            long leaseDuration);

    /**
     * Releases the lease for specific tableBuckets asynchronously.
     * 
     * <p> This is typically called when a client finishes reading a specific bucket (or a batch of buckets) 
     * but is still reading others under the same leaseId.
     *
     * <p>If {@code bucketsToRelease} contains all buckets under this leaseId, the lease 
     * itself will be removed.
     *
     * @param leaseId          The lease id.
     * @param bucketsToRelease The specific tableBuckets to release.
     */
    CompletableFuture<Void> releaseKvSnapshotLease(
            String leaseId, 
            Set<TableBucket> bucketsToRelease);

    /**
     * Drops the entire lease asynchronously.
     *
     * <p> All snapshots locked under this {@code leaseId} will be released immediately.
     * This is equivalent to calling {@link #releaseKvSnapshotLease} with all held buckets.
     *
     * @param leaseId The lease id to drop.
     */
    CompletableFuture<Void> dropKvSnapshotLease(String leaseId);
}

/**
 * A class to represent the result of acquire kv snapshot lease. It contains:
 *
 * <ul>
 *   <li>A map of unavailable snapshots. Such as the specify snapshotId is not exist for this table
 *       bucket.
 * </ul>
 *
 * @since 0.9
 */
@PublicEvolving
public class AcquireKvSnapshotLeaseResult {
    private final Map<TableBucket, Long> unavailableSnapshots;

    public AcquireKvSnapshotLeaseResult(Map<TableBucket, Long> unavailableSnapshots) {
        this.unavailableSnapshots = unavailableSnapshots;
    }

    /**
     * Returns the set of buckets that could not be locked (e.g., snapshot ID doesn't exist 
     * or has already been GC'ed).
     */
    public Map<TableBucket, Long> getUnavailableSnapshots() {
        return unavailableSnapshots;
    }
}

/**
 * Kv snapshot lease not exist exception.
 *
 * @since 0.9
 */
public class KvSnapshotLeaseNotExistException extends ApiException {
    private static final long serialVersionUID = 1L;

    public KvSnapshotLeaseNotExistException(String message) {
        super(message);
    }
}

message AcquireKvSnapshotLeaseRequest {
  required string lease_id = 1;
  required int64 lease_duration = 2;
  repeated PbKvSnapshotLeaseForTable tables_lease_req = 3;
}

message AcquireKvSnapshotLeaseResponse {
  repeated PbKvSnapshotLeaseForTable tables_lease_res = 3;
}

// util class
message PbKvSnapshotLeaseForTable {
  required int64 table_id = 1;
  repeated PbKvSnapshotLeaseForBucket buckets_lease_req = 2;
}

message PbKvSnapshotLeaseForBucket {
  optional int64 partition_id = 1;
  required int32 bucket_id = 2;
  required int64 snapshot_id = 3;
}

message ReleaseKvSnapshotLeaseRequest {
  required string lease_id = 1;
  repeated PbTable release_tables = 2;
}

message ReleaseKvSnapshotLeaseResponse {
}

// util class

message PbTable {
  required int64 table_id = 1;
  repeated PbBucket buckets = 2;
}

message PbBucket {
  optional int64 partition_id = 1;
  required int32 bucket_id = 2;
}

message DropKvSnapshotLeaseRequest {
  required string lease_id = 1;
}

message DropKvSnapshotLeaseResponse {
}

    // ------------------------------------------------------------------------------------------
    // ZNodes for Leases.
    // ------------------------------------------------------------------------------------------

    /** The root znode for leases. It will record all the info of fluss leases.
     *
	 * <p> Leases are a type of time-bound subscription. For example, a kv snapshot lease with lease duration:
	 *     Once the duration time is reached, the lease is automatically released, and the corresponding kv snapshot
	 *	   becomes eligible for deletion. However, as long as the lease remains active, the kv snapshot must not be deleted.
	 */
    public static final class LeasesNode {
        public static String path() {
            return "/leases";
        }
    }

    /** The root znode for kv snapshot leases. */
    public static final class KvSnapshotLeasesZNode {
        public static String path() {
            return LeasesNode.path() + "/kv_snapshot";
        }
    }

    /** The znode for kv snapshot lease. */
    public static final class KvSnapshotLeaseZNode {
        public static String path(String leaseId) {
            return KvSnapshotLeasesZNode.path() + "/" + leaseId;
        }

        public static byte[] encode(KvSnapshotLease kvSnapshotLease) {
            return JsonSerdeUtils.writeValueAsBytes(
                    kvSnapshotLease, kvSnapshotLeaseJsonSerde.INSTANCE);
        }

        public static KvSnapshotLease decode(byte[] json) {
            return JsonSerdeUtils.readValue(json, KvSnapshotLeaseJsonSerde.INSTANCE);
        }
    }

// 1. the content of zkNode
{
  "version": 1,
  "expiration_time": 1735538268434,
  "tables": [
   {
      "table_id": 1
      "lease_metadata_path": "{$remote_path}/lease/kv-snapshot/{$lease_id}/{$table_id}/{UUID}.metadata"  
    },
    {
      "table_id": 2
      "lease_metadata_path": "{$remote_path}/lease/kv-snapshot/{$lease_id}/{$table_id}/{UUID}.metadata"  
    }
  ]
}
 
// 2. the content of file: {$remote_path}/lease/kv-snapshot/{$lease_id}/{$table_id}/{UUID}.metadata
// 2.1 none-partition table

{
  "version": 1,
  "table_id": 1,
  "bucket_snapshots": [1, -1, 1, 2] // the value is a sorted list for bucket -> snapshotId, -1 means no lease for this bucket.
}


// 2.2 partition table

{
  "version": 1,
  "table_id": 2,
  "partition_snapshots": 
  [
    {
      "partition_id": 2001,
      "bucket_snapshots": [10, -1, 20, 30] // the value is a sorted list for bucket -> snapshotId, -1 means no lease for this bucket.

    },
    {
      "partition_id": 2002,
      "bucket_snapshots": [15, -1, 25, 35]
    }
  ]
}

public static final ConfigOption<Duration> KV_SNAPSHOT_LEASE_EXPIRATION_CHECK_INTERVAL =
   key("kv.snapshot.lease.expiration-check-interval")
   .durationType()
   .defaultValue(Duration.ofMinutes(10))
   .withDescription(
           "The interval to check the expiration of kv snapshot leases. "
                + "The default setting is 10 minutes.");

public static final ConfigOption<String> SCAN_KV_SNAPSHOT_LEASE_ID =
	ConfigOptions.key("scan.kv.snapshot.lease.id")
    .stringType()
    .defaultValue(String.valueOf(UUID.randomUUID()))
    .withDescription(
             "The lease id to ping kv snapshots. If set, the registered kv snapshots will not be deleted "
             		+ "until the consumer finished consuming all the snapshots or the lease duration time "
                    + "is reached. If not set, an UUID will be set.");

public static final ConfigOption<Duration> SCAN_KV_SNAPSHOT_LEASE_DURATION =
	ConfigOptions.key("scan.kv.snapshot.lease.duration")
		.durationType()
        .defaultValue(Duration.ofDays(1))
        .withDescription(
                "The time period how long to wait before expiring the kv snapshot lease to "
                      + "avoid kv snapshot blocking to delete. Default value is 1 day");

SELECT * FROM `kv_table`/*OPTIONAL('scan.kv.snapshot.lease.id'='lease1', 'scan.kv.snapshot.lease.duration'='1d')*/

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.fluss.flink.source.event;

import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.fluss.metadata.TableBucket;

import java.util.Objects;
import java.util.Set;

/** SourceEvent used to represent a Fluss table bucket has complete consume kv snapshot. */
public class FinishedKvSnapshotConsumeEvent implements SourceEvent {
    private static final long serialVersionUID = 1L;

    private final long checkpointId;
    /** The tableBucket set who finished consume kv snapshots. */
    private final Set<TableBucket> tableBuckets;

    public FinishedKvSnapshotConsumeEvent(long checkpointId, Set<TableBucket> tableBuckets) {
        this.checkpointId = checkpointId;
        this.tableBuckets = tableBuckets;
    }

    public long getCheckpointId() {
        return checkpointId;
    }

    public Set<TableBucket> getTableBuckets() {
        return tableBuckets;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        FinishedKvSnapshotConsumeEvent that = (FinishedKvSnapshotConsumeEvent) o;
        return checkpointId == that.checkpointId && tableBuckets.equals(that.tableBuckets);
    }

    @Override
    public int hashCode() {
        return Objects.hash(checkpointId, tableBuckets);
    }

    @Override
    public String toString() {
        return "FinishedKvSnapshotConsumeEvent{"
                + "checkpointId="
                + checkpointId
                + ", tableBuckets="
                + tableBuckets
                + '}';
    }
}

public class FlinkSourceReader<OUT>
        extends SingleThreadMultiplexSourceReaderBaseAdapter<
                RecordAndPos, OUT, SourceSplitBase, SourceSplitState> {

  ...

  /** the tableBuckets ignore to send FinishedKvSnapshotConsumeEvent as it already sending. */
  private final Set<TableBucket> ignoreBuckets;

  ...


  @Override
  public List<SourceSplitBase> snapshotState(long checkpointId) {
		Set<TableBucket> bucketsFinishedConsumeKvSnapshot = new HashSet<>();

        // do not modify this state.
        List<SourceSplitBase> sourceSplitBases = super.snapshotState(checkpointId);
        for (SourceSplitBase sourceSplitBase : sourceSplitBases) {
            TableBucket tableBucket = sourceSplitBase.getTableBucket();
            if (ignoreBuckets.contains(tableBucket)) {
                continue;
            }

            if (sourceSplitBase.isHybridSnapshotLogSplit()) {
                HybridSnapshotLogSplit hybridSnapshotLogSplit =
                        sourceSplitBase.asHybridSnapshotLogSplit();
                if (hybridSnapshotLogSplit.isSnapshotFinished()) {
                    bucketsFinishedConsumeKvSnapshot.add(tableBucket);
                }
            }
        }

        // report finished kv snapshot consume event.
        if (!bucketsFinishedConsumeKvSnapshot.isEmpty()) {
            context.sendSourceEventToCoordinator(
                    new FinishedKvSnapshotConsumeEvent(
                            checkpointId, bucketsFinishedConsumeKvSnapshot));
            // It won't be sent anymore in the future for this table bucket, but will be resent
            // after failover recovery as ignoreBuckets is cleared.
            ignoreBuckets.addAll(bucketsFinishedConsumeKvSnapshot);
        }

        return sourceSplitBases;
    }
}

public class FlinkSourceEnumerator
        implements SplitEnumerator<SourceSplitBase, SourceEnumeratorState> {
  ....

  private final String kvSnapshotLeaseId;
  private final ReadWriteLock lock = new ReentrantReadWriteLock();
    /** checkpointId -> tableBuckets who finished consume kv snapshots. */
  @GuardedBy("lock")
  private final TreeMap<Long, Set<TableBucket>> consumedKvSnapshotMap = new TreeMap<>();

  ...

  @Override
  public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
    if (sourceEvent instanceof ...)
    else if (sourceEvent instanceof FinishedKvSnapshotConsumeEvent) {
      FinishedKvSnapshotConsumeEvent event = (FinishedKvSnapshotConsumeEvent) sourceEvent;
      long checkpointId = event.getCheckpointId();
      event.getTableBuckets().forEach(tableBucket -> addConsumedBucket(checkpointId, tableBucket));
   }
  }

  @Override
  public void notifyCheckpointComplete(long checkpointId) throws Exception {
    // lower than this checkpoint id.
    Set<TableBucket> consumedKvSnapshots = getAndRemoveConsumedBucketsUpTo(checkpointId);

    // send request to fluss admin to release the kv snapshot lease.
    flussAdmin.releaseKvSnapshotLease(kvSnapshotLeaseId, consumedKvSnapshots).get();
  }

  /** Add bucket who has been consumed kv snapshot to the consumedKvSnapshotMap. */
  public void addConsumedBucket(long checkpointId, TableBucket tableBucket) {
        inWriteLock(
                lock,
                () -> {
                    consumedKvSnapshotMap
                            .computeIfAbsent(checkpointId, k -> new HashSet<>())
                            .add(tableBucket);
                });
  }

  /** Get and remove the buckets who have been consumed kv snapshot up to the checkpoint id. */
  public Set<TableBucket> getAndRemoveConsumedBucketsUpTo(long checkpointId) {
        return inWriteLock(
                lock,
                () -> {
                    NavigableMap<Long, Set<TableBucket>> toRemove =
                            consumedKvSnapshotMap.headMap(checkpointId, false);
                    Set<TableBucket> result = new HashSet<>();
                    for (Set<TableBucket> snapshots : toRemove.values()) {
                        result.addAll(snapshots);
                    }
                    toRemove.clear();
                    return result;
                });
  }
}

Call `fluss-catalog`.sys.drop_kv_snapshot_lease('snapsho-lease-1000');