[kv] Implement KvTablet RocksDB Lazy Open

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1903,6 +1903,23 @@ public class ConfigOptions {
                     .withDescription(
                             "The max fetch size for fetching log to apply to kv during recovering kv.");
 
+    // ------------------------------------------------------------------------
+    //  ConfigOptions for KV lazy open
+    // ------------------------------------------------------------------------
+
+    public static final ConfigOption<Boolean> KV_LAZY_OPEN_ENABLED =
+            key("kv.lazy-open.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Whether to enable KvTablet lazy open.");
+
+    public static final ConfigOption<Duration> KV_LAZY_OPEN_IDLE_TIMEOUT =
+            key("kv.lazy-open.idle-timeout")
+                    .durationType()
+                    .defaultValue(Duration.ofHours(24))
+                    .withDescription(
+                            "Idle time before an open KvTablet is eligible for release back to lazy state.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for metrics
     // ------------------------------------------------------------------------

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -147,6 +147,9 @@ public class MetricNames {
             "preWriteBufferTruncateAsDuplicatedPerSecond";
     public static final String KV_PRE_WRITE_BUFFER_TRUNCATE_AS_ERROR_RATE =
             "preWriteBufferTruncateAsErrorPerSecond";
+    public static final String KV_TABLET_OPEN_COUNT = "kvTabletOpenCount";
+    public static final String KV_TABLET_LAZY_COUNT = "kvTabletLazyCount";
+    public static final String KV_TABLET_FAILED_COUNT = "kvTabletFailedCount";
 
     // --------------------------------------------------------------------------------------------
     // RocksDB metrics

fluss-server/src/main/java/org/apache/fluss/server/kv/KvIdleReleaseController.java

@@ -0,0 +1,160 @@
+/*
+ * 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.server.kv;
+
+import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.utils.clock.Clock;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/**
+ * Periodically checks OPEN KvTablets and releases idle ones back to LAZY state. Uses LRU (least
+ * recently accessed) ordering. Operates directly on {@link KvTablet} — no dependency on Replica
+ * layer.
+ */
+public class KvIdleReleaseController implements Closeable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KvIdleReleaseController.class);
+
+    private final ScheduledExecutorService scheduler;
+    private final Supplier<Collection<KvTablet>> tabletSupplier;
+    private final Clock clock;
+
+    private final long checkIntervalMs;
+    private final long idleIntervalMs;
+
+    private volatile ScheduledFuture<?> scheduledTask;
+
+    public KvIdleReleaseController(
+            ScheduledExecutorService scheduler,
+            Supplier<Collection<KvTablet>> tabletSupplier,
+            Clock clock,
+            long checkIntervalMs,
+            long idleIntervalMs) {
+        this.scheduler = scheduler;
+        this.tabletSupplier = tabletSupplier;
+        this.clock = clock;
+        this.checkIntervalMs = checkIntervalMs;
+        this.idleIntervalMs = idleIntervalMs;
+    }
+
+    public void start() {
+        scheduledTask =
+                scheduler.scheduleWithFixedDelay(
+                        this::checkAndRelease,
+                        checkIntervalMs,
+                        checkIntervalMs,
+                        TimeUnit.MILLISECONDS);
+        LOG.info(
+                "KvIdleReleaseController started: checkInterval={}ms, idleInterval={}ms",
+                checkIntervalMs,
+                idleIntervalMs);
+    }
+
+    @VisibleForTesting
+    void checkAndRelease() {
+        try {
+            Collection<KvTablet> tablets = tabletSupplier.get();
+
+            long now = clock.milliseconds();
+
+            // Quick check: scan for any idle tablet before sorting
+            boolean hasIdle = false;
+            for (KvTablet t : tablets) {
+                if (now - t.getLastAccessTimestamp() > idleIntervalMs) {
+                    hasIdle = true;
+                    break;
+                }
+            }
+            if (!hasIdle) {
+                return;
+            }
+
+            // Sort by coldest first (LRU)
+            List<ReleaseCandidate> candidates =
+                    tablets.stream()
+                            .map(t -> new ReleaseCandidate(t, t.getLastAccessTimestamp()))
+                            .sorted(Comparator.comparingLong(c -> c.lastAccessTimestamp))
+                            .collect(Collectors.toList());
+
+            int idleCount = 0;
+            for (ReleaseCandidate c : candidates) {
+                if (now - c.lastAccessTimestamp > idleIntervalMs) {
+                    idleCount++;
+                } else {
+                    break;
+                }
+            }
+
+            LOG.info("Idle release round: open={}, idle={}", tablets.size(), idleCount);
+
+            int released = 0;
+            for (int i = 0; i < idleCount; i++) {
+                ReleaseCandidate candidate = candidates.get(i);
+                KvTablet tablet = candidate.tablet;
+
+                if (tablet.canRelease(idleIntervalMs, now)) {
+                    try {
+                        boolean success = tablet.releaseKv();
+                        if (success) {
+                            released++;
+                            LOG.debug(
+                                    "Released KvTablet for {} (idle {}ms)",
+                                    tablet.getTableBucket(),
+                                    now - candidate.lastAccessTimestamp);
+                        }
+                    } catch (Exception e) {
+                        LOG.warn("Failed to release KvTablet for {}", tablet.getTableBucket(), e);
+                    }
+                }
+            }
+
+            LOG.info("Idle release round complete: released={}/{}", released, idleCount);
+        } catch (Exception e) {
+            LOG.error("Error during idle release check", e);
+        }
+    }
+
+    @Override
+    public void close() {
+        if (scheduledTask != null) {
+            scheduledTask.cancel(false);
+        }
+    }
+
+    private static class ReleaseCandidate {
+        final KvTablet tablet;
+        final long lastAccessTimestamp;
+
+        ReleaseCandidate(KvTablet tablet, long lastAccessTimestamp) {
+            this.tablet = tablet;
+            this.lastAccessTimestamp = lastAccessTimestamp;
+        }
+    }
+}