[ISSUE #10270] Expose Pop RocksDB memtable size as configurable properties

Add writeBufferSize to MessageStoreConfig with a default of 32MB, replacing
the hardcoded value in createPopCFOptions(). maxWriteBufferNumber and
minWriteBufferNumberToMerge remain hardcoded as they rarely need tuning.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: lizhimins

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerRocksdbStore.java

@@ -47,10 +47,12 @@ public class PopConsumerRocksdbStore extends AbstractRocksDBStorage implements P
     private WriteOptions deleteOptions;
     protected ColumnFamilyHandle columnFamilyHandle;
     private final long blockCacheSize;
+    private final long writeBufferSize;
 
-    public PopConsumerRocksdbStore(String filePath, long blockCacheSize) {
+    public PopConsumerRocksdbStore(String filePath, long blockCacheSize, long writeBufferSize) {
         super(filePath);
         this.blockCacheSize = blockCacheSize;
+        this.writeBufferSize = writeBufferSize;
     }
 
     // https://www.cnblogs.com/renjc/p/rocksdb-class-db.html
@@ -85,8 +87,8 @@ protected boolean postLoad() {
             initOptions();
 
             // init column family here
-            ColumnFamilyOptions defaultOptions = RocksDBOptionsFactory.createPopCFOptions(blockCacheSize);
-            ColumnFamilyOptions popStateOptions = RocksDBOptionsFactory.createPopCFOptions(blockCacheSize);
+            ColumnFamilyOptions defaultOptions = RocksDBOptionsFactory.createPopCFOptions(blockCacheSize, writeBufferSize);
+            ColumnFamilyOptions popStateOptions = RocksDBOptionsFactory.createPopCFOptions(blockCacheSize, writeBufferSize);
             this.cfOptions.add(defaultOptions);
             this.cfOptions.add(popStateOptions);
 

broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java

@@ -99,7 +99,8 @@ public PopConsumerService(BrokerController brokerController) {
         this.consumerLockService = new PopConsumerLockService(TimeUnit.MINUTES.toMillis(2));
         this.popConsumerStore = new PopConsumerRocksdbStore(Paths.get(
             brokerController.getMessageStoreConfig().getStorePathRootDir(), ROCKSDB_DIRECTORY).toString(),
-            brokerController.getMessageStoreConfig().getPopRocksdbBlockCacheSize());
+            brokerController.getMessageStoreConfig().getPopRocksdbBlockCacheSize(),
+            brokerController.getMessageStoreConfig().getPopRocksdbWriteBufferSize());
         this.popConsumerCache = brokerConfig.isEnablePopBufferMerge() ? new PopConsumerCache(
             brokerController, this.popConsumerStore, this.consumerLockService, this::revive) : null;
 

broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerRocksdbStoreTest.java

@@ -66,7 +66,8 @@ public static PopConsumerRecord getConsumerRecord() {
     @Test
     public void rocksdbStoreWriteDeleteTest() {
         String filePath = getRandomStorePath();
-        PopConsumerKVStore consumerStore = new PopConsumerRocksdbStore(filePath, 256 * SizeUnit.MB);
+        PopConsumerKVStore consumerStore = new PopConsumerRocksdbStore(
+            filePath, 256 * SizeUnit.MB, 32 * SizeUnit.MB);
         Assert.assertEquals(filePath, consumerStore.getFilePath());
 
         consumerStore.start();
@@ -128,7 +129,8 @@ private long getDirectorySizeRecursive(File directory) {
     @Ignore
     @SuppressWarnings("ConstantValue")
     public void tombstoneDeletionTest() throws IllegalAccessException, NoSuchFieldException {
-        PopConsumerRocksdbStore rocksdbStore = new PopConsumerRocksdbStore(getRandomStorePath(), 256 * SizeUnit.MB);
+        PopConsumerRocksdbStore rocksdbStore = new PopConsumerRocksdbStore(
+            getRandomStorePath(), 256 * SizeUnit.MB, 32 * SizeUnit.MB);
         rocksdbStore.start();
 
         int iterCount = 1000 * 1000;

store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java

@@ -507,6 +507,8 @@ public class MessageStoreConfig {
 
     private long popRocksdbBlockCacheSize = 256 * SizeUnit.MB;
 
+    private long popRocksdbWriteBufferSize = 32 * SizeUnit.MB;
+
     /**
      * Flush RocksDB WAL frequency, aka, flush WAL every N write ops.
      */
@@ -541,6 +543,14 @@ public void setPopRocksdbBlockCacheSize(long popRocksdbBlockCacheSize) {
         this.popRocksdbBlockCacheSize = popRocksdbBlockCacheSize;
     }
 
+    public long getPopRocksdbWriteBufferSize() {
+        return popRocksdbWriteBufferSize;
+    }
+
+    public void setPopRocksdbWriteBufferSize(long popRocksdbWriteBufferSize) {
+        this.popRocksdbWriteBufferSize = popRocksdbWriteBufferSize;
+    }
+
     /**
      * Spin number in the retreat strategy of spin lock
      * Default is 1000

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

@@ -132,7 +132,7 @@ public static ColumnFamilyOptions createOffsetCFOptions() {
                 setInplaceUpdateSupport(true);
     }
 
-    public static ColumnFamilyOptions createPopCFOptions(long blockCacheSize) {
+    public static ColumnFamilyOptions createPopCFOptions(long blockCacheSize, long writeBufferSize) {
         BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig()
             .setFormatVersion(5)
             .setIndexType(IndexType.kBinarySearch)
@@ -160,7 +160,7 @@ public static ColumnFamilyOptions createPopCFOptions(long blockCacheSize) {
         //noinspection resource
         return new ColumnFamilyOptions()
             .setMaxWriteBufferNumber(4)
-            .setWriteBufferSize(128 * SizeUnit.MB)
+            .setWriteBufferSize(writeBufferSize)
             .setMinWriteBufferNumberToMerge(1)
             .setTableFormatConfig(blockBasedTableConfig)
             .setMemTableConfig(new SkipListMemTableConfig())