fix concurrency issue

Author: platinumhamburg

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

@@ -275,29 +275,44 @@ public void registerMetrics(BucketMetricGroup bucketMetricGroup) {
                 new MeterView(kvPreWriteBuffer.getTruncateAsErrorCount()));
 
         // Initialize RocksDB metrics reporter
-        if (rocksDBMetricsCollector == null && !isClosed) {
-            try {
-                // Get RocksDB Statistics from options container
-                org.rocksdb.Statistics statistics = rocksDBKv.getOptionsContainer().getStatistics();
-                if (statistics != null) {
-                    // Set server metric group in manager if not already set
-                    RocksDBMetricsManager.getInstance().setServerMetricGroup(serverMetricGroup);
-
-                    rocksDBMetricsCollector =
-                            new RocksDBMetricsCollector(
-                                    rocksDBKv.getDb(),
-                                    statistics,
-                                    tableBucket,
-                                    physicalPath.getTablePath(),
-                                    physicalPath.getPartitionName(),
-                                    rocksDBKv.getOptionsContainer());
+        // Use double-check locking pattern to ensure thread-safe initialization
+        if (rocksDBMetricsCollector == null) {
+            synchronized (this) {
+                // Double-check after acquiring lock
+                if (rocksDBMetricsCollector == null) {
+                    // Check if closed while holding lock
+                    boolean closed;
+                    synchronized (kvLock) {
+                        closed = isClosed;
+                    }
+                    if (!closed) {
+                        try {
+                            // Get RocksDB Statistics from options container
+                            org.rocksdb.Statistics statistics =
+                                    rocksDBKv.getOptionsContainer().getStatistics();
+                            if (statistics != null) {
+                                // Set server metric group in manager if not already set
+                                RocksDBMetricsManager.getInstance()
+                                        .setServerMetricGroup(serverMetricGroup);
+
+                                rocksDBMetricsCollector =
+                                        new RocksDBMetricsCollector(
+                                                rocksDBKv.getDb(),
+                                                statistics,
+                                                tableBucket,
+                                                physicalPath.getTablePath(),
+                                                physicalPath.getPartitionName(),
+                                                rocksDBKv.getOptionsContainer());
+                            }
+                        } catch (Exception e) {
+                            LOG.warn(
+                                    "Failed to initialize RocksDB metrics reporter for table {} bucket {}",
+                                    physicalPath,
+                                    tableBucket.getBucket(),
+                                    e);
+                        }
+                    }
                 }
-            } catch (Exception e) {
-                LOG.warn(
-                        "Failed to initialize RocksDB metrics reporter for table {} bucket {}",
-                        physicalPath,
-                        tableBucket.getBucket(),
-                        e);
             }
         }
     }

fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBMetricsCollector.java

@@ -329,30 +329,38 @@ private long computeBlockCachePinnedUsage() {
 
     @Override
     public void close() {
-        LOG.info(
-                "Closing RocksDB metrics collector for table {} bucket {}.",
-                tablePath,
-                tableBucket.getBucket());
-
-        // Set cleanup in progress flag to prevent new operations
-        cleanupInProgress = true;
-
-        // Set closed flag to prevent new operations
-        closed = true;
+        // Use synchronized block to ensure thread-safe close operation
+        synchronized (this) {
+            // Check if already closed to prevent double closing
+            if (closed) {
+                return;
+            }
 
-        // Unregister from global collector
-        if (registered) {
-            try {
-                RocksDBMetricsManager.getInstance().unregisterCollector(this);
-            } catch (Exception e) {
-                LOG.warn("Error unregistering collector from manager: {}", e.getMessage());
-            } finally {
-                registered = false;
+            LOG.info(
+                    "Closing RocksDB metrics collector for table {} bucket {}.",
+                    tablePath,
+                    tableBucket.getBucket());
+
+            // Set cleanup in progress flag to prevent new operations
+            cleanupInProgress = true;
+
+            // Set closed flag to prevent new operations
+            closed = true;
+
+            // Unregister from global collector
+            if (registered) {
+                try {
+                    RocksDBMetricsManager.getInstance().unregisterCollector(this);
+                } catch (Exception e) {
+                    LOG.warn("Error unregistering collector from manager: {}", e.getMessage());
+                } finally {
+                    registered = false;
+                }
             }
-        }
 
-        // Signal cleanup completion
-        cleanupLatch.countDown();
+            // Signal cleanup completion
+            cleanupLatch.countDown();
+        }
     }
 
     public TableBucket getTableBucket() {

fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBMetricsManager.java

@@ -154,16 +154,14 @@ public void unregisterCollector(RocksDBMetricsCollector collector) {
 
         synchronized (collectorLock) {
             String key = createCollectorKey(collector);
-            synchronized (key.intern()) {
-                RocksDBMetricsCollector removed = collectors.remove(key);
-                if (removed != null) {
-                    LOG.debug(
-                            "Unregistered RocksDB metrics collector for {}, remaining collectors: {}",
-                            key,
-                            collectors.size());
-                } else {
-                    LOG.debug("Collector {} was not found in registry", key);
-                }
+            RocksDBMetricsCollector removed = collectors.remove(key);
+            if (removed != null) {
+                LOG.debug(
+                        "Unregistered RocksDB metrics collector for {}, remaining collectors: {}",
+                        key,
+                        collectors.size());
+            } else {
+                LOG.debug("Collector {} was not found in registry", key);
             }
         }
     }

fluss-server/src/main/java/org/apache/fluss/server/kv/rocksdb/RocksDBResourceContainer.java

@@ -303,7 +303,7 @@ private ColumnFamilyOptions setColumnFamilyOptionsFromConfigurableOptions(
 
         // Create explicit cache instance instead of using setBlockCacheSize
         long blockCacheSize = internalGetOption(ConfigOptions.KV_BLOCK_CACHE_SIZE).getBytes();
-        if (blockCacheSize > 0) {
+        if (blockCacheSize > 0 && blockCache != null) {
             blockBasedTableConfig.setBlockCache(blockCache);
         }