[kv] full options support for shared block cache

Author: platinumhamburg

fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java

@@ -1548,6 +1548,25 @@ public class ConfigOptions {
                     .defaultValue(MemorySize.parse("4gb"))
                     .withDescription("The size of the shared block cache if enabled.");
 
+    public static final ConfigOption<Boolean> KV_SHARED_BLOCK_CACHE_STRICT_CAPACITY_LIMIT =
+            key("kv.rocksdb.shared-block-cache.strict-capacity-limit")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Whether to strictly limit the shared block cache capacity.");
+
+    public static final ConfigOption<Integer> KV_SHARED_BLOCK_CACHE_NUM_SHARD_BITS =
+            key("kv.rocksdb.shared-block-cache.num-shard-bits")
+                    .intType()
+                    .defaultValue(-1)
+                    .withDescription(
+                            "Number of bits for shard count (8 means 256 shards) in the shared block cache.");
+
+    public static final ConfigOption<Double> KV_SHARED_BLOCK_CACHE_HIGH_PRI_POOL_RATIO =
+            key("kv.rocksdb.shared-block-cache.high-pri-pool-ratio")
+                    .doubleType()
+                    .defaultValue(0.0)
+                    .withDescription("Ratio of high priority pool in the shared block cache.");
+
     public static final ConfigOption<Boolean> KV_USE_BLOOM_FILTER =
             key("kv.rocksdb.use-bloom-filter")
                     .booleanType()

fluss-server/src/main/java/com/alibaba/fluss/server/kv/rocksdb/RocksDBSharedResource.java

@@ -47,8 +47,7 @@
  *       released
  * </ul>
  *
- * <p>Closeable state determination: When reference count is 0, it's in closeable state, no
- * additional state variables needed.
+ * <p>Closeable state determination: When reference count is 0, it's in closeable state.
  *
  * <p>Note: This class is thread-safe and can be used concurrently across multiple threads.
  */
@@ -145,8 +144,6 @@ public int release() {
             LOG.debug("Released RocksDBSharedResource, reference count: {}", newCount);
 
             if (newCount == 0) {
-                // When reference count is 0, don't release resources, just enter closeable state
-                LOG.debug("RocksDBSharedResource transitioned to closeable state");
                 // Wake up any waiting close() method
                 lock.notifyAll();
             } else if (newCount < 0) {
@@ -242,6 +239,12 @@ private void createSharedBlockCache() {
         }
 
         long cacheSize = configuration.get(ConfigOptions.KV_SHARED_BLOCK_CACHE_SIZE).getBytes();
+        int cacheNumShardBits =
+                configuration.get(ConfigOptions.KV_SHARED_BLOCK_CACHE_NUM_SHARD_BITS);
+        boolean strictCapacityLimit =
+                configuration.get(ConfigOptions.KV_SHARED_BLOCK_CACHE_STRICT_CAPACITY_LIMIT);
+        double highPriPoolRatio =
+                configuration.get(ConfigOptions.KV_SHARED_BLOCK_CACHE_HIGH_PRI_POOL_RATIO);
 
         // Load RocksDB native library if needed, this operation is idempotent
         RocksDB.loadLibrary();
@@ -252,9 +255,15 @@ private void createSharedBlockCache() {
         // - numShardBits: number of bits for shard count (8 means 256 shards)
         // - strictCapacityLimit: whether to strictly limit capacity
         // - highPriPoolRatio: ratio of high priority pool
-        sharedBlockCache = new LRUCache(cacheSize, 8, true, 0.5);
-
-        LOG.info("Created shared block cache with size: {} bytes", cacheSize);
+        sharedBlockCache =
+                new LRUCache(cacheSize, cacheNumShardBits, strictCapacityLimit, highPriPoolRatio);
+
+        LOG.info(
+                "Created shared block cache with size: {} bytes, numShardBits: {}, strictCapacityLimit: {}, highPriPoolRatio: {}",
+                cacheSize,
+                cacheNumShardBits,
+                strictCapacityLimit,
+                highPriPoolRatio);
     }
 
     /** Close shared resources. */