[ISSUE #10334] Support native CqCompactionFilter with cross-platform JNI shim

.github/workflows/maven.yaml

@@ -27,7 +27,13 @@ jobs:
           distribution: "corretto"
           cache: "maven"
       - name: Build with Maven
-        run: mvn -B package --file pom.xml
+        shell: bash
+        run: |
+          JACOCO_FLAG=""
+          if [ "${{ matrix.os }}" != "ubuntu-latest" ]; then
+            JACOCO_FLAG="-Djacoco.skip=true"
+          fi
+          mvn -B package --file pom.xml $JACOCO_FLAG
 
       - name: Upload Auth JVM crash logs
         if: failure()

common/pom.xml

@@ -113,8 +113,8 @@
             <artifactId>rocketmq-logback-classic</artifactId>
         </dependency>
         <dependency>
-            <groupId>org.apache.rocketmq</groupId>
-            <artifactId>rocketmq-rocksdb</artifactId>
+            <groupId>org.rocksdb</groupId>
+            <artifactId>rocksdbjni</artifactId>
         </dependency>
     </dependencies>
 </project>

common/src/main/java/org/apache/rocketmq/common/config/AbstractRocksDBStorage.java

@@ -427,22 +427,35 @@ protected void manualCompactionDefaultCfRange(CompactRangeOptions compactRangeOp
         if (!hold()) {
             return;
         }
-        long s1 = System.currentTimeMillis();
+        long before = getEstimateNumKeys();
+        long startMs = System.currentTimeMillis();
         boolean result = true;
         try {
-            LOGGER.info("manualCompaction Start. {}", this.dbPath);
+            LOGGER.info("ManualCompaction started, dbPath={}, estimateNumKeys={}", this.dbPath, before);
             this.db.compactRange(this.defaultCFHandle, null, null, compactRangeOptions);
         } catch (RocksDBException e) {
             result = false;
             scheduleReloadRocksdb(e);
-            LOGGER.error("manualCompaction Failed. {}, {}", this.dbPath, getStatusError(e));
+            LOGGER.error("ManualCompaction failed, dbPath={}, error={}", this.dbPath, getStatusError(e));
         } finally {
             release();
-            LOGGER.info("manualCompaction End. {}, rt: {}(ms), result: {}", this.dbPath, System.currentTimeMillis() - s1, result);
+            long after = getEstimateNumKeys();
+            long elapsed = System.currentTimeMillis() - startMs;
+            String ratio = before > 0 ? String.format("%.1f", (1.0 - (double) after / before) * 100) : "0.0";
+            LOGGER.info("ManualCompaction finished, dbPath={}, elapsed={}ms, success={}, before={}, after={}, reduced={}%",
+                this.dbPath, elapsed, result, before, after, ratio);
         }
     }
 
-    protected void manualCompaction(long minPhyOffset, final CompactRangeOptions compactRangeOptions) {
+    private long getEstimateNumKeys() {
+        try {
+            return this.db.getLongProperty(this.defaultCFHandle, "rocksdb.estimate-num-keys");
+        } catch (RocksDBException e) {
+            return -1L;
+        }
+    }
+
+    protected void manualCompaction(final CompactRangeOptions compactRangeOptions) {
         this.manualCompactionThread.submit(new Runnable() {
             @Override
             public void run() {

pom.xml

@@ -139,7 +139,7 @@
         <opentelemetry-exporter-prometheus.version>1.47.0-alpha</opentelemetry-exporter-prometheus.version>
         <jul-to-slf4j.version>2.0.6</jul-to-slf4j.version>
         <s3.version>2.20.29</s3.version>
-        <rocksdb.version>1.0.6</rocksdb.version>
+        <rocksdbjni.version>8.4.4</rocksdbjni.version>
         <jackson-databind.version>2.13.4.2</jackson-databind.version>
         <sofa-jraft.version>1.3.14</sofa-jraft.version>
 
@@ -761,9 +761,9 @@
                 <version>${slf4j-api.version}</version>
             </dependency>
             <dependency>
-                <groupId>org.apache.rocketmq</groupId>
-                <artifactId>rocketmq-rocksdb</artifactId>
-                <version>${rocksdb.version}</version>
+                <groupId>org.rocksdb</groupId>
+                <artifactId>rocksdbjni</artifactId>
+                <version>${rocksdbjni.version}</version>
             </dependency>
             <dependency>
                 <groupId>io.github.aliyunmq</groupId>

store/pom.xml

@@ -71,5 +71,9 @@
             <groupId>io.github.aliyunmq</groupId>
             <artifactId>rocketmq-shaded-slf4j-api-bridge</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.rocksdb</groupId>
+            <artifactId>rocksdbjni</artifactId>
+        </dependency>
     </dependencies>
 </project>

store/src/main/java/org/apache/rocketmq/store/rocksdb/ConsumeQueueRocksDBStorage.java

@@ -21,10 +21,14 @@
 import java.util.List;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.config.AbstractRocksDBStorage;
+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.ColumnFamilyDescriptor;
 import org.rocksdb.ColumnFamilyHandle;
 import org.rocksdb.ColumnFamilyOptions;
+import org.rocksdb.FlushOptions;
 import org.rocksdb.ReadOptions;
 import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
@@ -33,13 +37,13 @@
 
 public class ConsumeQueueRocksDBStorage extends AbstractRocksDBStorage {
 
+    private static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKSDB_LOGGER_NAME);
+
     public static final byte[] OFFSET_COLUMN_FAMILY = "offset".getBytes(StandardCharsets.UTF_8);
 
     private final MessageStore messageStore;
     private volatile ColumnFamilyHandle offsetCFHandle;
 
-    private ConsumeQueueCompactionFilterFactory compactionFilterFactory;
-
     public ConsumeQueueRocksDBStorage(final MessageStore messageStore, final String dbPath) {
         super(dbPath);
         this.messageStore = messageStore;
@@ -67,20 +71,27 @@ protected boolean postLoad() {
 
             final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>();
 
-            this.compactionFilterFactory = new ConsumeQueueCompactionFilterFactory(messageStore::getMinPhyOffset);
-
-            ColumnFamilyOptions cqCfOptions = RocksDBOptionsFactory.createCQCFOptions(this.messageStore, this.compactionFilterFactory);
+            ColumnFamilyOptions cqCfOptions = RocksDBOptionsFactory.createCQCFOptions(this.messageStore);
             this.cfOptions.add(cqCfOptions);
             cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, cqCfOptions));
 
             ColumnFamilyOptions offsetCfOptions = RocksDBOptionsFactory.createOffsetCFOptions();
             this.cfOptions.add(offsetCfOptions);
             cfDescriptors.add(new ColumnFamilyDescriptor(OFFSET_COLUMN_FAMILY, offsetCfOptions));
+
+            if (CqCompactionFilterJni.isLoaded()) {
+                CqCompactionFilterJni.createAndSetFilter(cqCfOptions);
+                CqCompactionFilterJni.setMinPhyOffset(messageStore.getMinPhyOffset());
+                log.info("CqCompactionFilter created and set, minPhyOffset: {}", messageStore.getMinPhyOffset());
+            } else {
+                log.warn("CqCompactionFilterJni native library not loaded, compaction filter will not be installed");
+            }
+
             open(cfDescriptors);
             this.defaultCFHandle = cfHandles.get(0);
             this.offsetCFHandle = cfHandles.get(1);
         } catch (final Exception e) {
-            LOGGER.error("postLoad Failed. {}", this.dbPath, e);
+            log.error("postLoad Failed. {}", this.dbPath, e);
             return false;
         }
         return true;
@@ -91,11 +102,6 @@ protected void preShutdown() {
         if (this.offsetCFHandle != null) {
             this.offsetCFHandle.close();
         }
-
-        if (this.compactionFilterFactory != null) {
-            this.compactionFilterFactory.close();
-        }
-
     }
 
     public byte[] getCQ(final byte[] keyBytes) throws RocksDBException {
@@ -116,10 +122,13 @@ public void batchPut(final WriteBatch batch) throws RocksDBException {
     }
 
     public void manualCompaction(final long minPhyOffset) {
+        if (CqCompactionFilterJni.isLoaded()) {
+            CqCompactionFilterJni.setMinPhyOffset(minPhyOffset);
+        }
         try {
-            manualCompaction(minPhyOffset, this.compactRangeOptions);
+            super.manualCompaction(this.compactRangeOptions);
         } catch (Exception e) {
-            LOGGER.error("manualCompaction Failed. minPhyOffset: {}", minPhyOffset, e);
+            log.error("manualCompaction Failed. minPhyOffset: {}", minPhyOffset, e);
         }
     }
 
@@ -130,4 +139,41 @@ public RocksIterator seekOffsetCF() {
     public ColumnFamilyHandle getOffsetCFHandle() {
         return this.offsetCFHandle;
     }
+
+    /**
+     * Synchronously trigger compaction with an updated compaction filter threshold.
+     * This method updates the native compaction filter's minPhyOffset and then
+     * performs a full compaction on the default column family.
+     */
+    public void triggerCompactionSync(long minPhyOffset) throws RocksDBException {
+        if (CqCompactionFilterJni.isLoaded()) {
+            CqCompactionFilterJni.setMinPhyOffset(minPhyOffset);
+        }
+        db.compactRange(this.defaultCFHandle);
+    }
+
+    /**
+     * Flush all memtables to SST files.
+     */
+    public void flushAll() throws RocksDBException {
+        try (FlushOptions flushOpts = new FlushOptions()) {
+            flushOpts.setWaitForFlush(true);
+            flush(flushOpts);
+        }
+    }
+
+    /**
+     * Count all entries in the default column family by iterating. O(N), use only in tests.
+     */
+    public long countEntries() {
+        long count = 0;
+        try (RocksIterator iter = db.newIterator(this.defaultCFHandle)) {
+            iter.seekToFirst();
+            while (iter.isValid()) {
+                count++;
+                iter.next();
+            }
+        }
+        return count;
+    }
 }