(rocketmq) branch develop updated: [ISSUE #9680] Improve RocksDB compaction filter factory resource management (#9681)

[lt...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ltamber pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new c3c6832f35 [ISSUE #9680] Improve RocksDB compaction filter factory resource management (#9681)
c3c6832f35 is described below

commit c3c6832f355cf589d9e1033db55490eda87b0951
Author: rongtong <ji...@mails.ucas.ac.cn>
AuthorDate: Thu Sep 11 14:15:02 2025 +0800

    [ISSUE #9680] Improve RocksDB compaction filter factory resource management (#9681)
    
    * feat: improve rocksdb compaction filter factory resource management
    
    - Refactor ConsumeQueueCompactionFilterFactory to use LongSupplier instead of MessageStore
    - Add proper resource cleanup in ConsumeQueueRocksDBStorage.preShutdown()
    - Update RocksDBOptionsFactory to accept external compaction filter factory
    - Optimize write buffer size for CQ rocksdb performance
    
    This change improves resource management and reduces memory leaks by:
    1. Decoupling compaction filter from MessageStore reference
    2. Ensuring proper cleanup of native resources
    3. Making compaction filter factory lifecycle manageable
    
    * Polish the code
    
    * Fix consumeQueueCompactionFilterFactory not use
---
 .../rocksdb/ConsumeQueueCompactionFilterFactory.java   | 10 +++++-----
 .../store/rocksdb/ConsumeQueueRocksDBStorage.java      | 18 +++++++++++++++---
 .../rocketmq/store/rocksdb/RocksDBOptionsFactory.java  |  5 +++--
 3 files changed, 23 insertions(+), 10 deletions(-)

diff --git a/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueCompactionFilterFactory.java b/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueCompactionFilterFactory.java
index aa796c4d39..f19fb9e203 100644
--- a/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueCompactionFilterFactory.java
+++ b/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueCompactionFilterFactory.java
@@ -16,20 +16,20 @@
  */
 package org.apache.rocketmq.store.rocksdb;
 
+import java.util.function.LongSupplier;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.org.slf4j.Logger;
 import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
-import org.apache.rocketmq.store.MessageStore;
 import org.rocksdb.AbstractCompactionFilter;
 import org.rocksdb.AbstractCompactionFilterFactory;
 import org.rocksdb.RemoveConsumeQueueCompactionFilter;
 
 public class ConsumeQueueCompactionFilterFactory extends AbstractCompactionFilterFactory<RemoveConsumeQueueCompactionFilter> {
     private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKSDB_LOGGER_NAME);
-    private final MessageStore messageStore;
+    private final LongSupplier minPhyOffsetSupplier;
 
-    public ConsumeQueueCompactionFilterFactory(final MessageStore messageStore) {
-        this.messageStore = messageStore;
+    public ConsumeQueueCompactionFilterFactory(final LongSupplier minPhyOffsetSupplier) {
+        this.minPhyOffsetSupplier = minPhyOffsetSupplier;
     }
 
     @Override
@@ -39,7 +39,7 @@ public class ConsumeQueueCompactionFilterFactory extends AbstractCompactionFilte
 
     @Override
     public RemoveConsumeQueueCompactionFilter createCompactionFilter(final AbstractCompactionFilter.Context context) {
-        long minPhyOffset = this.messageStore.getMinPhyOffset();
+        long minPhyOffset = this.minPhyOffsetSupplier.getAsLong();
         LOGGER.info("manualCompaction minPhyOffset: {}, isFull: {}, isManual: {}",
                 minPhyOffset, context.isFullCompaction(), context.isManualCompaction());
         return new RemoveConsumeQueueCompactionFilter(minPhyOffset);
diff --git a/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueRocksDBStorage.java b/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueRocksDBStorage.java
index b04aeab6bd..4392283c67 100644
--- a/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueRocksDBStorage.java
+++ b/store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueRocksDBStorage.java
@@ -38,6 +38,8 @@ public class ConsumeQueueRocksDBStorage extends AbstractRocksDBStorage {
     private final MessageStore messageStore;
     private volatile ColumnFamilyHandle offsetCFHandle;
 
+    private ConsumeQueueCompactionFilterFactory compactionFilterFactory;
+
     public ConsumeQueueRocksDBStorage(final MessageStore messageStore, final String dbPath) {
         super(dbPath);
         this.messageStore = messageStore;
@@ -65,7 +67,9 @@ public class ConsumeQueueRocksDBStorage extends AbstractRocksDBStorage {
 
             final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>();
 
-            ColumnFamilyOptions cqCfOptions = RocksDBOptionsFactory.createCQCFOptions(this.messageStore);
+            this.compactionFilterFactory = new ConsumeQueueCompactionFilterFactory(messageStore::getMinPhyOffset);
+
+            ColumnFamilyOptions cqCfOptions = RocksDBOptionsFactory.createCQCFOptions(this.messageStore, this.compactionFilterFactory);
             this.cfOptions.add(cqCfOptions);
             cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cqCfOptions));
 
@@ -84,7 +88,14 @@ public class ConsumeQueueRocksDBStorage extends AbstractRocksDBStorage {
 
     @Override
     protected void preShutdown() {
-        this.offsetCFHandle.close();
+        if (this.offsetCFHandle != null) {
+            this.offsetCFHandle.close();
+        }
+
+        if (this.compactionFilterFactory != null) {
+            this.compactionFilterFactory.close();
+        }
+
     }
 
     public byte[] getCQ(final byte[] keyBytes) throws RocksDBException {
@@ -95,7 +106,8 @@ public class ConsumeQueueRocksDBStorage extends AbstractRocksDBStorage {
         return get(this.offsetCFHandle, this.totalOrderReadOptions, keyBytes);
     }
 
-    public List<byte[]> multiGet(final List<ColumnFamilyHandle> cfhList, final List<byte[]> keys) throws RocksDBException {
+    public List<byte[]> multiGet(final List<ColumnFamilyHandle> cfhList,
+        final List<byte[]> keys) throws RocksDBException {
         return multiGet(this.totalOrderReadOptions, cfhList, keys);
     }
 
diff --git a/store/src/main/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactory.java b/store/src/main/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactory.java
index 5687d6a222..e365326c76 100644
--- a/store/src/main/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactory.java
+++ b/store/src/main/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactory.java
@@ -41,7 +41,8 @@ import org.rocksdb.util.SizeUnit;
 
 public class RocksDBOptionsFactory {
 
-    public static ColumnFamilyOptions createCQCFOptions(final MessageStore messageStore) {
+    public static ColumnFamilyOptions createCQCFOptions(final MessageStore messageStore,
+        ConsumeQueueCompactionFilterFactory consumeQueueCompactionFilterFactory) {
         BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig().
                 setFormatVersion(5).
                 setIndexType(IndexType.kBinarySearch).
@@ -92,7 +93,7 @@ public class RocksDBOptionsFactory {
                 setTargetFileSizeBase(256 * SizeUnit.MB).
                 setTargetFileSizeMultiplier(2).
                 setMergeOperator(new StringAppendOperator()).
-                setCompactionFilterFactory(new ConsumeQueueCompactionFilterFactory(messageStore)).
+                setCompactionFilterFactory(consumeQueueCompactionFilterFactory).
                 setReportBgIoStats(true).
                 setOptimizeFiltersForHits(true);
     }