opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-08-30 18:10:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: multiple patches during long running tests for LMQ over RocksDB (#8915)

Browse Source 
* fix: multiple patches during long running tests for LMQ over RocksDB

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

* fix: fix a bug in RocksGroupCommitService; remove RocksDBConsumeQueueStore#findConsumeQueueMap override

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

* fix: async fsync on RocksDB WAL flush

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

* fix: use a dedicated thread to flush and sync RocksDB WAL

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

* fix: trigger WAL rolling according to estimated WAL file size

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

* chore: add doc, explaining config RocksDB instance flush/sync strategy

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

* fix: data-version should be per table

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

* fix: test case: RocksdbTransferOffsetAndCqTest

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>

---------

Signed-off-by: Li Zhanhui <lizhanhui@gmail.com>
This commit is contained in:
Zhanhui Li
2024-12-02 10:07:54 +08:00
committed by GitHub
parent ea1228a30a
commit 9a891f1d49
18 changed files with 472 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+1 -1
View File
@@ -344,7 +344,7 @@ public class BrokerController {
this.brokerStatsManager = messageStoreConfig.isEnableLmq() ? new LmqBrokerStatsManager(this.brokerConfig) : new BrokerStatsManager(this.brokerConfig.getBrokerClusterName(), this.brokerConfig.isEnableDetailStat());
this.broadcastOffsetManager = new BroadcastOffsetManager(this);
if (ConfigManagerVersion.V2.getVersion().equals(brokerConfig.getConfigManagerVersion())) {
this.configStorage = new ConfigStorage(messageStoreConfig.getStorePathRootDir());
this.configStorage = new ConfigStorage(messageStoreConfig);
this.topicConfigManager = new TopicConfigManagerV2(this, configStorage);
this.subscriptionGroupManager = new SubscriptionGroupManagerV2(this, configStorage);
this.consumerOffsetManager = new ConsumerOffsetManagerV2(this, configStorage);
broker/src/main/java/org/apache/rocketmq/broker/config/v2/ConfigHelper.java
+2 -2
View File
@@ -64,7 +64,7 @@ public class ConfigHelper {
return Optional.empty();
}
public static void stampDataVersion(WriteBatch writeBatch, DataVersion dataVersion, long stateMachineVersion)
public static void stampDataVersion(WriteBatch writeBatch, TableId table, DataVersion dataVersion, long stateMachineVersion)
throws RocksDBException {
// Increase data version
dataVersion.nextVersion(stateMachineVersion);
@@ -75,7 +75,7 @@ public class ConfigHelper {
ByteBuf valueBuf = AbstractRocksDBStorage.POOLED_ALLOCATOR.buffer(Long.BYTES * 3);
try {
keyBuf.writeByte(TablePrefix.TABLE.getValue());
keyBuf.writeShort(TableId.CONSUMER_OFFSET.getValue());
keyBuf.writeShort(table.getValue());
keyBuf.writeByte(RecordPrefix.DATA_VERSION.getValue());
keyBuf.writeBytes(ConfigStorage.DATA_VERSION_KEY_BYTES);
valueBuf.writeLong(dataVersion.getStateVersion());
broker/src/main/java/org/apache/rocketmq/broker/config/v2/ConfigStorage.java
+151 -5
View File
@@ -16,17 +16,29 @@
*/
package org.apache.rocketmq.broker.config.v2;
import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.buffer.PooledByteBufAllocatorMetric;
import io.netty.util.internal.PlatformDependent;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.config.AbstractRocksDBStorage;
import org.apache.rocketmq.common.config.ConfigHelper;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.rocksdb.ColumnFamilyDescriptor;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.FlushOptions;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
@@ -43,8 +55,48 @@ public class ConfigStorage extends AbstractRocksDBStorage {
public static final String DATA_VERSION_KEY = "data_version";
public static final byte[] DATA_VERSION_KEY_BYTES = DATA_VERSION_KEY.getBytes(StandardCharsets.UTF_8);
public ConfigStorage(String storePath) {
super(storePath + File.separator + "config" + File.separator + "rdb");
private final ScheduledExecutorService scheduledExecutorService;
/**
* Number of write ops since previous flush.
*/
private final AtomicInteger writeOpsCounter;
private final AtomicLong estimateWalFileSize = new AtomicLong(0L);
private final MessageStoreConfig messageStoreConfig;
private final FlushSyncService flushSyncService;
public ConfigStorage(MessageStoreConfig messageStoreConfig) {
super(messageStoreConfig.getStorePathRootDir() + File.separator + "config" + File.separator + "rdb");
this.messageStoreConfig = messageStoreConfig;
ThreadFactory threadFactory = new ThreadFactoryBuilder()
.setDaemon(true)
.setNameFormat("config-storage-%d")
.build();
scheduledExecutorService = new ScheduledThreadPoolExecutor(1, threadFactory);
writeOpsCounter = new AtomicInteger(0);
this.flushSyncService = new FlushSyncService();
this.flushSyncService.setDaemon(true);
}
private void statNettyMemory() {
PooledByteBufAllocatorMetric metric = AbstractRocksDBStorage.POOLED_ALLOCATOR.metric();
LOGGER.info("Netty Memory Usage: {}", metric);
}
@Override
public synchronized boolean start() {
boolean started = super.start();
if (started) {
scheduledExecutorService.scheduleWithFixedDelay(() -> statRocksdb(LOGGER), 1, 10, TimeUnit.SECONDS);
scheduledExecutorService.scheduleWithFixedDelay(this::statNettyMemory, 10, 10, TimeUnit.SECONDS);
this.flushSyncService.start();
} else {
LOGGER.error("Failed to start config storage");
}
return started;
}
@Override
@@ -58,7 +110,7 @@ public class ConfigStorage extends AbstractRocksDBStorage {
initOptions();
List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>();
ColumnFamilyOptions defaultOptions = ConfigHelper.createConfigOptions();
ColumnFamilyOptions defaultOptions = ConfigHelper.createConfigColumnFamilyOptions();
this.cfOptions.add(defaultOptions);
cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, defaultOptions));
@@ -66,7 +118,7 @@ public class ConfigStorage extends AbstractRocksDBStorage {
open(cfDescriptors);
this.defaultCFHandle = cfHandles.get(0);
} catch (final Exception e) {
} catch (Exception e) {
AbstractRocksDBStorage.LOGGER.error("postLoad Failed. {}", this.dbPath, e);
return false;
}
@@ -75,7 +127,8 @@ public class ConfigStorage extends AbstractRocksDBStorage {
@Override
protected void preShutdown() {
scheduledExecutorService.shutdown();
flushSyncService.shutdown();
}
protected void initOptions() {
@@ -105,6 +158,12 @@ public class ConfigStorage extends AbstractRocksDBStorage {
public void write(WriteBatch writeBatch) throws RocksDBException {
db.write(ableWalWriteOptions, writeBatch);
accountWriteOps(writeBatch.getDataSize());
}
private void accountWriteOps(long dataSize) {
writeOpsCounter.incrementAndGet();
estimateWalFileSize.addAndGet(dataSize);
}
public RocksIterator iterate(ByteBuffer beginKey, ByteBuffer endKey) {
@@ -125,4 +184,91 @@ public class ConfigStorage extends AbstractRocksDBStorage {
return iterator;
}
}
/**
* RocksDB writes contain 3 stages: application memory buffer --> OS Page Cache --> Disk.
* Given that we are having DBOptions::manual_wal_flush, we need to manually call DB::FlushWAL and DB::SyncWAL
* Note: DB::FlushWAL(true) will internally call DB::SyncWAL.
* <p>
* See <a href="https://rocksdb.org/blog/2017/08/25/flushwal.html">Flush And Sync WAL</a>
*/
class FlushSyncService extends ServiceThread {
private long lastSyncTime = 0;
private static final long MAX_SYNC_INTERVAL_IN_MILLIS = 100;
private final Stopwatch stopwatch = Stopwatch.createUnstarted();
private final FlushOptions flushOptions = new FlushOptions();
@Override
public String getServiceName() {
return "FlushSyncService";
}
@Override
public void run() {
flushOptions.setAllowWriteStall(false);
flushOptions.setWaitForFlush(true);
log.info("{} service started", this.getServiceName());
while (!this.isStopped()) {
try {
this.waitForRunning(10);
this.flushAndSyncWAL(false);
} catch (Exception e) {
log.warn("{} service has exception. ", this.getServiceName(), e);
}
}
try {
flushAndSyncWAL(true);
} catch (Exception e) {
log.warn("{} raised an exception while performing flush-and-sync WAL on exit",
this.getServiceName(), e);
}
flushOptions.close();
log.info("{} service end", this.getServiceName());
}
private void flushAndSyncWAL(boolean onExit) throws RocksDBException {
int writeOps = writeOpsCounter.get();
if (0 == writeOps) {
// No write ops to flush
return;
}
/*
* Normally, when MemTables become full then immutable, RocksDB threads will automatically flush them to L0
* SST files. The use case here is different: the MemTable may never get full and immutable given that the
* volume of data involved is relatively small. Further, we are constantly modifying the key-value pairs and
* generating WAL entries. The WAL file size can grow up to dozens of gigabytes without manual triggering of
* flush.
*/
if (ConfigStorage.this.estimateWalFileSize.get() >= messageStoreConfig.getRocksdbWalFileRollingThreshold()) {
ConfigStorage.this.flush(flushOptions);
estimateWalFileSize.set(0L);
}
// Flush and Sync WAL if we have committed enough writes
if (writeOps >= messageStoreConfig.getRocksdbFlushWalFrequency() || onExit) {
stopwatch.reset().start();
ConfigStorage.this.db.flushWal(true);
long elapsed = stopwatch.stop().elapsed(TimeUnit.MILLISECONDS);
writeOpsCounter.getAndAdd(-writeOps);
lastSyncTime = System.currentTimeMillis();
LOGGER.debug("Flush and Sync WAL of RocksDB[{}] costs {}ms, write-ops={}", dbPath, elapsed, writeOps);
return;
}
// Flush and Sync WAL if some writes are out there for a period of time
long elapsedTime = System.currentTimeMillis() - lastSyncTime;
if (elapsedTime > MAX_SYNC_INTERVAL_IN_MILLIS) {
stopwatch.reset().start();
ConfigStorage.this.db.flushWal(true);
long elapsed = stopwatch.stop().elapsed(TimeUnit.MILLISECONDS);
LOGGER.debug("Flush and Sync WAL of RocksDB[{}] costs {}ms, write-ops={}", dbPath, elapsed, writeOps);
writeOpsCounter.getAndAdd(-writeOps);
lastSyncTime = System.currentTimeMillis();
}
}
}
}
broker/src/main/java/org/apache/rocketmq/broker/config/v2/ConsumerOffsetManagerV2.java
+4 -4
View File
@@ -97,7 +97,7 @@ public class ConsumerOffsetManagerV2 extends ConsumerOffsetManager {
// TODO: we have to make a copy here as WriteBatch lacks ByteBuffer API here
writeBatch.deleteRange(ConfigHelper.readBytes(beginKey), ConfigHelper.readBytes(endKey));
long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.CONSUMER_OFFSET, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
} catch (RocksDBException e) {
LOG.error("Failed to removeConsumerOffset, topicAtGroup={}", topicAtGroup, e);
@@ -138,7 +138,7 @@ public class ConsumerOffsetManagerV2 extends ConsumerOffsetManager {
writeBatch.deleteRange(ConfigHelper.readBytes(beginKey), ConfigHelper.readBytes(endKey));
MessageStore messageStore = brokerController.getMessageStore();
long stateMachineVersion = messageStore != null ? messageStore.getStateMachineVersion() : 0;
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.CONSUMER_OFFSET, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
} catch (RocksDBException e) {
LOG.error("Failed to consumer offsets by group={}", group, e);
@@ -194,7 +194,7 @@ public class ConsumerOffsetManagerV2 extends ConsumerOffsetManager {
writeBatch.put(keyBuf.nioBuffer(), valueBuf.nioBuffer());
MessageStore messageStore = brokerController.getMessageStore();
long stateMachineVersion = messageStore != null ? messageStore.getStateMachineVersion() : 0;
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.CONSUMER_OFFSET, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
} catch (RocksDBException e) {
LOG.error("Failed to commit consumer offset", e);
@@ -394,7 +394,7 @@ public class ConsumerOffsetManagerV2 extends ConsumerOffsetManager {
try (WriteBatch writeBatch = new WriteBatch()) {
writeBatch.put(keyBuf.nioBuffer(), valueBuf.nioBuffer());
long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.PULL_OFFSET, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
} catch (RocksDBException e) {
LOG.error("Failed to commit pull offset. group={}, topic={}, queueId={}, offset={}",
broker/src/main/java/org/apache/rocketmq/broker/config/v2/SubscriptionGroupManagerV2.java
+4 -2
View File
@@ -137,8 +137,10 @@ public class SubscriptionGroupManagerV2 extends SubscriptionGroupManager {
try (WriteBatch writeBatch = new WriteBatch()) {
writeBatch.put(keyBuf.nioBuffer(), valueBuf.nioBuffer());
long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.SUBSCRIPTION_GROUP, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
// fdatasync on core metadata change
persist();
} catch (RocksDBException e) {
log.error("update subscription group config error", e);
} finally {
@@ -163,7 +165,7 @@ public class SubscriptionGroupManagerV2 extends SubscriptionGroupManager {
try (WriteBatch writeBatch = new WriteBatch()) {
writeBatch.delete(ConfigHelper.readBytes(keyBuf));
long stateMachineVersion = brokerController.getMessageStore().getStateMachineVersion();
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.SUBSCRIPTION_GROUP, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
} catch (RocksDBException e) {
log.error("Failed to remove subscription group config by group-name={}", groupName, e);
broker/src/main/java/org/apache/rocketmq/broker/config/v2/TopicConfigManagerV2.java
+4 -2
View File
@@ -151,8 +151,10 @@ public class TopicConfigManagerV2 extends TopicConfigManager {
try (WriteBatch writeBatch = new WriteBatch()) {
writeBatch.put(keyBuf.nioBuffer(), valueBuf.nioBuffer());
long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.TOPIC, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
// fdatasync on core metadata change
this.persist();
} catch (RocksDBException e) {
log.error("Failed to update topic config", e);
} finally {
@@ -167,7 +169,7 @@ public class TopicConfigManagerV2 extends TopicConfigManager {
try (WriteBatch writeBatch = new WriteBatch()) {
writeBatch.delete(keyBuf.nioBuffer());
long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
ConfigHelper.stampDataVersion(writeBatch, dataVersion, stateMachineVersion);
ConfigHelper.stampDataVersion(writeBatch, TableId.TOPIC, dataVersion, stateMachineVersion);
configStorage.write(writeBatch);
} catch (RocksDBException e) {
log.error("Failed to delete topic config by topicName={}", topicName, e);
broker/src/test/java/org/apache/rocketmq/broker/config/v2/ConsumerOffsetManagerV2Test.java
+18 -1
View File
@@ -23,6 +23,7 @@ import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -44,6 +45,8 @@ public class ConsumerOffsetManagerV2Test {
@Mock
private BrokerController controller;
private MessageStoreConfig messageStoreConfig;
@Rule
public TemporaryFolder tf = new TemporaryFolder();
@@ -60,7 +63,9 @@ public class ConsumerOffsetManagerV2Test {
Mockito.doReturn(brokerConfig).when(controller).getBrokerConfig();
File configStoreDir = tf.newFolder();
configStorage = new ConfigStorage(configStoreDir.getAbsolutePath());
messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setStorePathRootDir(configStoreDir.getAbsolutePath());
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
consumerOffsetManagerV2 = new ConsumerOffsetManagerV2(controller, configStorage);
}
@@ -84,7 +89,9 @@ public class ConsumerOffsetManagerV2Test {
consumerOffsetManagerV2.getOffsetTable().clear();
Assert.assertEquals(-1L, consumerOffsetManagerV2.queryOffset(group, topic, queueId));
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
consumerOffsetManagerV2 = new ConsumerOffsetManagerV2(controller, configStorage);
consumerOffsetManagerV2.load();
Assert.assertEquals(queueOffset, consumerOffsetManagerV2.queryOffset(group, topic, queueId));
}
@@ -106,7 +113,9 @@ public class ConsumerOffsetManagerV2Test {
configStorage.shutdown();
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
consumerOffsetManagerV2 = new ConsumerOffsetManagerV2(controller, configStorage);
consumerOffsetManagerV2.load();
Assert.assertEquals(queueOffset, consumerOffsetManagerV2.queryOffset(group, topic, queueId));
}
@@ -129,7 +138,9 @@ public class ConsumerOffsetManagerV2Test {
configStorage.shutdown();
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
consumerOffsetManagerV2 = new ConsumerOffsetManagerV2(controller, configStorage);
consumerOffsetManagerV2.load();
Assert.assertEquals(queueOffset, consumerOffsetManagerV2.queryPullOffset(group, topic, queueId));
}
@@ -157,7 +168,10 @@ public class ConsumerOffsetManagerV2Test {
Assert.assertEquals(queueOffset, consumerOffsetManagerV2.queryOffset(group, topic2, queueId));
configStorage.shutdown();
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
consumerOffsetManagerV2 = new ConsumerOffsetManagerV2(controller, configStorage);
consumerOffsetManagerV2.load();
Assert.assertEquals(-1L, consumerOffsetManagerV2.queryOffset(group, topic, queueId));
Assert.assertEquals(queueOffset, consumerOffsetManagerV2.queryOffset(group, topic2, queueId));
@@ -184,7 +198,10 @@ public class ConsumerOffsetManagerV2Test {
Assert.assertEquals(-1L, consumerOffsetManagerV2.queryOffset(group, topic2, queueId));
configStorage.shutdown();
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
consumerOffsetManagerV2 = new ConsumerOffsetManagerV2(controller, configStorage);
consumerOffsetManagerV2.load();
Assert.assertEquals(-1L, consumerOffsetManagerV2.queryOffset(group, topic, queueId));
Assert.assertEquals(-1L, consumerOffsetManagerV2.queryOffset(group, topic2, queueId));
broker/src/test/java/org/apache/rocketmq/broker/config/v2/SubscriptionGroupManagerV2Test.java
+14 -1
View File
@@ -25,6 +25,7 @@ import org.apache.rocketmq.remoting.protocol.subscription.GroupRetryPolicy;
import org.apache.rocketmq.remoting.protocol.subscription.GroupRetryPolicyType;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -38,6 +39,9 @@ import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class SubscriptionGroupManagerV2Test {
private MessageStoreConfig messageStoreConfig;
private ConfigStorage configStorage;
private SubscriptionGroupManagerV2 subscriptionGroupManagerV2;
@@ -68,7 +72,9 @@ public class SubscriptionGroupManagerV2Test {
Mockito.doReturn(1L).when(messageStore).getStateMachineVersion();
File configStoreDir = tf.newFolder();
configStorage = new ConfigStorage(configStoreDir.getAbsolutePath());
messageStoreConfig = new MessageStoreConfig();
messageStoreConfig.setStorePathRootDir(configStoreDir.getAbsolutePath());
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
subscriptionGroupManagerV2 = new SubscriptionGroupManagerV2(controller, configStorage);
}
@@ -98,7 +104,10 @@ public class SubscriptionGroupManagerV2Test {
subscriptionGroupManagerV2.getSubscriptionGroupTable().clear();
configStorage.shutdown();
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
subscriptionGroupManagerV2 = new SubscriptionGroupManagerV2(controller, configStorage);
subscriptionGroupManagerV2.load();
found = subscriptionGroupManagerV2.findSubscriptionGroupConfig(subscriptionGroupConfig.getGroupName());
Assert.assertEquals(subscriptionGroupConfig, found);
@@ -132,7 +141,11 @@ public class SubscriptionGroupManagerV2Test {
Assert.assertNull(found);
configStorage.shutdown();
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
subscriptionGroupManagerV2 = new SubscriptionGroupManagerV2(controller, configStorage);
subscriptionGroupManagerV2.load();
found = subscriptionGroupManagerV2.findSubscriptionGroupConfig(subscriptionGroupConfig.getGroupName());
Assert.assertNull(found);
broker/src/test/java/org/apache/rocketmq/broker/config/v2/TopicConfigManagerV2Test.java
+19 -6
View File
@@ -23,6 +23,7 @@ import java.io.IOException;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.junit.After;
import org.junit.Assert;
@@ -35,17 +36,19 @@ import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(value = MockitoJUnitRunner.class)
public class TopicConfigManagerV2Test {
private ConfigStorage configStorage;
private MessageStoreConfig messageStoreConfig;
private TopicConfigManagerV2 topicConfigManagerV2;
private ConfigStorage configStorage;
@Mock
private BrokerController controller;
@Mock
private MessageStore messageStore;
@Rule
public TemporaryFolder tf = new TemporaryFolder();
@@ -61,17 +64,22 @@ public class TopicConfigManagerV2Test {
BrokerConfig brokerConfig = new BrokerConfig();
Mockito.doReturn(brokerConfig).when(controller).getBrokerConfig();
MessageStoreConfig messageStoreConfig = new MessageStoreConfig();
messageStoreConfig = new MessageStoreConfig();
Mockito.doReturn(messageStoreConfig).when(controller).getMessageStoreConfig();
Mockito.doReturn(messageStore).when(controller).getMessageStore();
File configStoreDir = tf.newFolder();
configStorage = new ConfigStorage(configStoreDir.getAbsolutePath());
messageStoreConfig.setStorePathRootDir(configStoreDir.getAbsolutePath());
configStorage = new ConfigStorage(messageStoreConfig);
configStorage.start();
topicConfigManagerV2 = new TopicConfigManagerV2(controller, configStorage);
}
@Test
public void testUpdateTopicConfig() {
TopicConfigManagerV2 topicConfigManagerV2 = new TopicConfigManagerV2(controller, configStorage);
topicConfigManagerV2.load();
TopicConfig topicConfig = new TopicConfig();
String topicName = "T1";
topicConfig.setTopicName(topicName);
@@ -86,7 +94,9 @@ public class TopicConfigManagerV2Test {
topicConfigManagerV2.getTopicConfigTable().clear();
configStorage = new ConfigStorage(messageStoreConfig);
Assert.assertTrue(configStorage.start());
topicConfigManagerV2 = new TopicConfigManagerV2(controller, configStorage);
Assert.assertTrue(topicConfigManagerV2.load());
TopicConfig loaded = topicConfigManagerV2.selectTopicConfig(topicName);
@@ -111,12 +121,15 @@ public class TopicConfigManagerV2Test {
topicConfig.setWriteQueueNums(4);
topicConfig.setOrder(true);
topicConfig.setTopicSysFlag(4);
TopicConfigManagerV2 topicConfigManagerV2 = new TopicConfigManagerV2(controller, configStorage);
topicConfigManagerV2.updateTopicConfig(topicConfig);
topicConfigManagerV2.removeTopicConfig(topicName);
Assert.assertFalse(topicConfigManagerV2.containsTopic(topicName));
Assert.assertTrue(configStorage.shutdown());
configStorage = new ConfigStorage(messageStoreConfig);
Assert.assertTrue(configStorage.start());
topicConfigManagerV2 = new TopicConfigManagerV2(controller, configStorage);
Assert.assertTrue(topicConfigManagerV2.load());
Assert.assertFalse(topicConfigManagerV2.containsTopic(topicName));
}
broker/src/test/java/org/apache/rocketmq/broker/offset/RocksdbTransferOffsetAndCqTest.java
+13 -3
View File
@@ -23,11 +23,11 @@ import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import org.apache.commons.collections.MapUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.config.v1.RocksDBConsumerOffsetManager;
import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.store.DefaultMessageStore;
@@ -38,6 +38,7 @@ import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.ConsumeQueueStoreInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.awaitility.Awaitility;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -135,6 +136,7 @@ public class RocksdbTransferOffsetAndCqTest {
}
RocksDBMessageStore kvStore = defaultMessageStore.getRocksDBMessageStore();
ConsumeQueueStoreInterface store = kvStore.getConsumeQueueStore();
store.start();
ConsumeQueueInterface rocksdbCq = defaultMessageStore.getRocksDBMessageStore().findConsumeQueue(topic, queueId);
ConsumeQueueInterface fileCq = defaultMessageStore.findConsumeQueue(topic, queueId);
for (int i = 0; i < 200; i++) {
@@ -142,13 +144,21 @@ public class RocksdbTransferOffsetAndCqTest {
fileCq.putMessagePositionInfoWrapper(request);
store.putMessagePositionInfoWrapper(request);
}
Awaitility.await()
.pollInterval(100, TimeUnit.MILLISECONDS)
.atMost(3, TimeUnit.SECONDS)
.until(() -> rocksdbCq.getMaxOffsetInQueue() == 200);
Pair<CqUnit, Long> unit = rocksdbCq.getCqUnitAndStoreTime(100);
Pair<CqUnit, Long> unit1 = fileCq.getCqUnitAndStoreTime(100);
Assert.assertTrue(unit.getObject1().getPos() == unit1.getObject1().getPos());
Assert.assertEquals(unit.getObject1().getPos(), unit1.getObject1().getPos());
}
/**
* No need to skip macOS platform.
* @return true if some platform is NOT a good fit for this test case.
*/
private boolean notToBeExecuted() {
return MixAll.isMac();
return false;
}
}
common/src/main/java/org/apache/rocketmq/common/config/AbstractRocksDBStorage.java
+18 -16
View File
@@ -121,14 +121,16 @@ public abstract class AbstractRocksDBStorage {
this.writeOptions = new WriteOptions();
this.writeOptions.setSync(false);
this.writeOptions.setDisableWAL(true);
this.writeOptions.setNoSlowdown(true);
// https://github.com/facebook/rocksdb/wiki/Write-Stalls
this.writeOptions.setNoSlowdown(false);
}
protected void initAbleWalWriteOptions() {
this.ableWalWriteOptions = new WriteOptions();
this.ableWalWriteOptions.setSync(false);
this.ableWalWriteOptions.setDisableWAL(false);
this.ableWalWriteOptions.setNoSlowdown(true);
// https://github.com/facebook/rocksdb/wiki/Write-Stalls
this.ableWalWriteOptions.setNoSlowdown(false);
}
protected void initReadOptions() {
@@ -363,7 +365,7 @@ public abstract class AbstractRocksDBStorage {
}
if (postLoad()) {
this.loaded = true;
LOGGER.info("start OK. {}", this.dbPath);
LOGGER.info("RocksDB[{}] starts OK", this.dbPath);
this.closed = false;
return true;
} else {
@@ -560,7 +562,15 @@ public abstract class AbstractRocksDBStorage {
public void statRocksdb(Logger logger) {
try {
// Log Memory Usage
String blockCacheMemUsage = this.db.getProperty("rocksdb.block-cache-usage");
String indexesAndFilterBlockMemUsage = this.db.getProperty("rocksdb.estimate-table-readers-mem");
String memTableMemUsage = this.db.getProperty("rocksdb.cur-size-all-mem-tables");
String blocksPinnedByIteratorMemUsage = this.db.getProperty("rocksdb.block-cache-pinned-usage");
logger.info("RocksDB Memory Usage: BlockCache: {}, IndexesAndFilterBlock: {}, MemTable: {}, BlocksPinnedByIterator: {}",
blockCacheMemUsage, indexesAndFilterBlockMemUsage, memTableMemUsage, blocksPinnedByIteratorMemUsage);
// Log file metadata by level
List<LiveFileMetaData> liveFileMetaDataList = this.getCompactionStatus();
if (liveFileMetaDataList == null || liveFileMetaDataList.isEmpty()) {
return;
@@ -570,21 +580,13 @@ public abstract class AbstractRocksDBStorage {
StringBuilder sb = map.computeIfAbsent(metaData.level(), k -> new StringBuilder(256));
sb.append(new String(metaData.columnFamilyName(), StandardCharsets.UTF_8)).append(SPACE).
append(metaData.fileName()).append(SPACE).
append("s: ").append(metaData.size()).append(SPACE).
append("a: ").append(metaData.numEntries()).append(SPACE).
append("r: ").append(metaData.numReadsSampled()).append(SPACE).
append("d: ").append(metaData.numDeletions()).append(SPACE).
append(metaData.beingCompacted()).append("\n");
append("file-size: ").append(metaData.size()).append(SPACE).
append("number-of-entries: ").append(metaData.numEntries()).append(SPACE).
append("file-read-times: ").append(metaData.numReadsSampled()).append(SPACE).
append("deletions: ").append(metaData.numDeletions()).append(SPACE).
append("being-compacted: ").append(metaData.beingCompacted()).append("\n");
}
map.forEach((key, value) -> logger.info("level: {}\n{}", key, value.toString()));
String blockCacheMemUsage = this.db.getProperty("rocksdb.block-cache-usage");
String indexesAndFilterBlockMemUsage = this.db.getProperty("rocksdb.estimate-table-readers-mem");
String memTableMemUsage = this.db.getProperty("rocksdb.cur-size-all-mem-tables");
String blocksPinnedByIteratorMemUsage = this.db.getProperty("rocksdb.block-cache-pinned-usage");
logger.info("MemUsage. blockCache: {}, indexesAndFilterBlock: {}, MemTable: {}, blocksPinnedByIterator: {}",
blockCacheMemUsage, indexesAndFilterBlockMemUsage, memTableMemUsage, blocksPinnedByIteratorMemUsage);
} catch (Exception ignored) {
}
}
common/src/main/java/org/apache/rocketmq/common/config/ConfigHelper.java
+20 -12
View File
@@ -38,7 +38,7 @@ import org.rocksdb.WALRecoveryMode;
import org.rocksdb.util.SizeUnit;
public class ConfigHelper {
public static ColumnFamilyOptions createConfigOptions() {
public static ColumnFamilyOptions createConfigColumnFamilyOptions() {
BlockBasedTableConfig blockBasedTableConfig = new BlockBasedTableConfig().
setFormatVersion(5).
setIndexType(IndexType.kBinarySearch).
@@ -46,7 +46,7 @@ public class ConfigHelper {
setBlockSize(32 * SizeUnit.KB).
setFilterPolicy(new BloomFilter(16, false)).
// Indicating if we'd put index/filter blocks to the block cache.
setCacheIndexAndFilterBlocks(false).
setCacheIndexAndFilterBlocks(true).
setCacheIndexAndFilterBlocksWithHighPriority(true).
setPinL0FilterAndIndexBlocksInCache(false).
setPinTopLevelIndexAndFilter(true).
@@ -54,9 +54,8 @@ public class ConfigHelper {
setWholeKeyFiltering(true);
ColumnFamilyOptions options = new ColumnFamilyOptions();
return options.setMaxWriteBufferNumber(2).
// MemTable size, MemTable(cache) -> immutable MemTable(cache) -> SST(disk)
setWriteBufferSize(8 * SizeUnit.MB).
return options.setMaxWriteBufferNumber(4).
setWriteBufferSize(64 * SizeUnit.MB).
setMinWriteBufferNumberToMerge(1).
setTableFormatConfig(blockBasedTableConfig).
setMemTableConfig(new SkipListMemTableConfig()).
@@ -67,17 +66,17 @@ public class ConfigHelper {
setLevel0SlowdownWritesTrigger(8).
setLevel0StopWritesTrigger(12).
// The target file size for compaction.
setTargetFileSizeBase(64 * SizeUnit.MB).
setTargetFileSizeBase(64 * SizeUnit.MB).
setTargetFileSizeMultiplier(2).
// The upper-bound of the total size of L1 files in bytes
setMaxBytesForLevelBase(256 * SizeUnit.MB).
setMaxBytesForLevelBase(256 * SizeUnit.MB).
setMaxBytesForLevelMultiplier(2).
setMergeOperator(new StringAppendOperator()).
setInplaceUpdateSupport(true);
}
public static DBOptions createConfigDBOptions() {
//Turn based on https://github.com/facebook/rocksdb/wiki/RocksDB-Tuning-Guide
// Tune based on https://github.com/facebook/rocksdb/wiki/RocksDB-Tuning-Guide
// and http://gitlab.alibaba-inc.com/aloha/aloha/blob/branch_2_5_0/jstorm-core/src/main/java/com/alibaba/jstorm/cache/rocksdb/RocksDbOptionsFactory.java
DBOptions options = new DBOptions();
Statistics statistics = new Statistics();
@@ -86,10 +85,20 @@ public class ConfigHelper {
setDbLogDir(getDBLogDir()).
setInfoLogLevel(InfoLogLevel.INFO_LEVEL).
setWalRecoveryMode(WALRecoveryMode.SkipAnyCorruptedRecords).
/*
* We use manual flush to achieve desired balance between reliability and performance:
* for metadata that matters, including {topic, subscription}-config changes, each write incurs a
* flush-and-sync to ensure reliability; for {commit, pull}-offset advancements, group-flush are offered for
* every N(configurable, 1024 by default) writes or aging of writes, similar to OS page-cache flush
* mechanism.
*/
setManualWalFlush(true).
setMaxTotalWalSize(500 * SizeUnit.MB).
setWalSizeLimitMB(0).
setWalTtlSeconds(0).
// This option takes effect only when we have multiple column families
// https://github.com/facebook/rocksdb/issues/4180
// setMaxTotalWalSize(1024 * SizeUnit.MB).
setDbWriteBufferSize(128 * SizeUnit.MB).
setBytesPerSync(SizeUnit.MB).
setWalBytesPerSync(SizeUnit.MB).
setCreateIfMissing(true).
setCreateMissingColumnFamilies(true).
setMaxOpenFiles(-1).
@@ -99,7 +108,6 @@ public class ConfigHelper {
setAllowConcurrentMemtableWrite(false).
setStatistics(statistics).
setStatsDumpPeriodSec(600).
setAtomicFlush(true).
setMaxBackgroundJobs(32).
setMaxSubcompactions(4).
setParanoidChecks(true).
common/src/main/java/org/apache/rocketmq/common/config/ConfigRocksDBStorage.java
+1 -1
View File
@@ -70,7 +70,7 @@ public class ConfigRocksDBStorage extends AbstractRocksDBStorage {
final List<ColumnFamilyDescriptor> cfDescriptors = new ArrayList<>();
ColumnFamilyOptions defaultOptions = ConfigHelper.createConfigOptions();
ColumnFamilyOptions defaultOptions = ConfigHelper.createConfigColumnFamilyOptions();
this.cfOptions.add(defaultOptions);
cfDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, defaultOptions));
cfDescriptors.add(new ColumnFamilyDescriptor(KV_DATA_VERSION_COLUMN_FAMILY_NAME, defaultOptions));
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+10 -6
View File
@@ -326,22 +326,26 @@ public class CommitLog implements Swappable {
boolean checkDupInfo = this.defaultMessageStore.getMessageStoreConfig().isDuplicationEnable();
final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
if (!mappedFiles.isEmpty()) {
// Began to recover from the last third file
int index = mappedFiles.size() - 3;
if (index < 0) {
index = 0;
int index = mappedFiles.size() - 1;
while (index > 0) {
MappedFile mappedFile = mappedFiles.get(index);
if (mappedFile.getFileFromOffset() <= maxPhyOffsetOfConsumeQueue) {
// It's safe to recover from this mapped file
break;
}
index--;
}
// TODO: Discuss if we need to load more commit-log mapped files into memory.
MappedFile mappedFile = mappedFiles.get(index);
ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
long processOffset = mappedFile.getFileFromOffset();
long mappedFileOffset = 0;
long lastValidMsgPhyOffset = this.getConfirmOffset();
// normal recover doesn't require dispatching
boolean doDispatch = false;
while (true) {
DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover, checkDupInfo);
int size = dispatchRequest.getMsgSize();
boolean doDispatch = dispatchRequest.getCommitLogOffset() > maxPhyOffsetOfConsumeQueue;
// Normal data
if (dispatchRequest.isSuccess() && size > 0) {
lastValidMsgPhyOffset = processOffset + mappedFileOffset;
store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
+24
View File
@@ -23,6 +23,7 @@ import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.StoreType;
import org.apache.rocketmq.store.queue.BatchConsumeQueue;
import org.rocksdb.CompressionType;
import org.rocksdb.util.SizeUnit;
public class MessageStoreConfig {
@@ -444,6 +445,13 @@ public class MessageStoreConfig {
private String rocksdbCompressionType = CompressionType.LZ4_COMPRESSION.getLibraryName();
/**
* Flush RocksDB WAL frequency, aka, flush WAL every N write ops.
*/
private int rocksdbFlushWalFrequency = 1024;
private long rocksdbWalFileRollingThreshold = SizeUnit.GB;
public String getRocksdbCompressionType() {
return rocksdbCompressionType;
}
@@ -1902,6 +1910,22 @@ public class MessageStoreConfig {
this.bottomMostCompressionTypeForConsumeQueueStore = bottomMostCompressionTypeForConsumeQueueStore;
}
public int getRocksdbFlushWalFrequency() {
return rocksdbFlushWalFrequency;
}
public void setRocksdbFlushWalFrequency(int rocksdbFlushWalFrequency) {
this.rocksdbFlushWalFrequency = rocksdbFlushWalFrequency;
}
public long getRocksdbWalFileRollingThreshold() {
return rocksdbWalFileRollingThreshold;
}
public void setRocksdbWalFileRollingThreshold(long rocksdbWalFileRollingThreshold) {
this.rocksdbWalFileRollingThreshold = rocksdbWalFileRollingThreshold;
}
public int getSpinLockCollisionRetreatOptimalDegree() {
return spinLockCollisionRetreatOptimalDegree;
}
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java
+62 -29
View File
@@ -30,12 +30,14 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nonnull;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.BoundaryType;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageConst;
@@ -84,6 +86,10 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
private final OffsetInitializer offsetInitializer;
private final RocksGroupCommitService groupCommitService;
private final AtomicReference<ServiceState> serviceState = new AtomicReference<>(ServiceState.CREATE_JUST);
public RocksDBConsumeQueueStore(DefaultMessageStore messageStore) {
super(messageStore);
@@ -93,6 +99,7 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
this.rocksDBConsumeQueueOffsetTable = new RocksDBConsumeQueueOffsetTable(rocksDBConsumeQueueTable, rocksDBStorage, messageStore);
this.offsetInitializer = new OffsetInitializerRocksDBImpl(this);
this.groupCommitService = new RocksGroupCommitService(this);
this.cqBBPairList = new ArrayList<>(16);
this.offsetBBPairList = new ArrayList<>(DEFAULT_BYTE_BUFFER_CAPACITY);
for (int i = 0; i < DEFAULT_BYTE_BUFFER_CAPACITY; i++) {
@@ -123,14 +130,17 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
@Override
public void start() {
log.info("RocksDB ConsumeQueueStore start!");
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
this.rocksDBStorage.statRocksdb(ROCKSDB_LOG);
}, 10, this.messageStoreConfig.getStatRocksDBCQIntervalSec(), TimeUnit.SECONDS);
if (serviceState.compareAndSet(ServiceState.CREATE_JUST, ServiceState.RUNNING)) {
log.info("RocksDB ConsumeQueueStore start!");
this.groupCommitService.start();
this.scheduledExecutorService.scheduleAtFixedRate(() -> {
this.rocksDBStorage.statRocksdb(ROCKSDB_LOG);
}, 10, this.messageStoreConfig.getStatRocksDBCQIntervalSec(), TimeUnit.SECONDS);
this.scheduledExecutorService.scheduleWithFixedDelay(() -> {
cleanDirty(messageStore.getTopicConfigs().keySet());
}, 10, this.messageStoreConfig.getCleanRocksDBDirtyCQIntervalMin(), TimeUnit.MINUTES);
this.scheduledExecutorService.scheduleWithFixedDelay(() -> {
cleanDirty(messageStore.getTopicConfigs().keySet());
}, 10, this.messageStoreConfig.getCleanRocksDBDirtyCQIntervalMin(), TimeUnit.MINUTES);
}
}
private void cleanDirty(final Set<String> existTopicSet) {
@@ -165,18 +175,23 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
@Override
public void recover() {
// ignored
start();
}
@Override
public boolean recoverConcurrently() {
start();
return true;
}
@Override
public boolean shutdown() {
this.scheduledExecutorService.shutdown();
return shutdownInner();
if (serviceState.compareAndSet(ServiceState.RUNNING, ServiceState.SHUTDOWN_ALREADY)) {
this.groupCommitService.shutdown();
this.scheduledExecutorService.shutdown();
return shutdownInner();
}
return true;
}
private boolean shutdownInner() {
@@ -188,23 +203,25 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
if (null == request) {
return;
}
// We are taking advantage of Atomic Flush, this operation is purely memory-based.
// batch and cache in Java heap does not make sense, instead, we should put the metadata into RocksDB immediately
// to optimized overall end-to-end latency.
putMessagePosition(request);
try {
groupCommitService.putRequest(request);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public void putMessagePosition(DispatchRequest request) throws RocksDBException {
public void putMessagePosition(List<DispatchRequest> requests) throws RocksDBException {
final int maxRetries = 30;
for (int i = 0; i < maxRetries; i++) {
if (putMessagePosition0(request)) {
if (putMessagePosition0(requests)) {
if (this.isCQError) {
this.messageStore.getRunningFlags().clearLogicsQueueError();
this.isCQError = false;
}
return;
} else {
ERROR_LOG.warn("{} put cq Failed. retryTime: {}", i);
ERROR_LOG.warn("Put cq Failed. retryTime: {}", i);
try {
Thread.sleep(100);
} catch (InterruptedException ignored) {
@@ -219,34 +236,43 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
throw new RocksDBException("put CQ Failed");
}
private boolean putMessagePosition0(DispatchRequest request) {
private boolean putMessagePosition0(List<DispatchRequest> requests) {
if (!this.rocksDBStorage.hold()) {
return false;
}
try (WriteBatch writeBatch = new WriteBatch()) {
final int size = requests.size();
if (size == 0) {
return true;
}
long maxPhyOffset = 0;
DispatchEntry entry = DispatchEntry.from(request);
dispatch(entry, writeBatch);
dispatchLMQ(request, writeBatch);
for (int i = size - 1; i >= 0; i--) {
final DispatchRequest request = requests.get(i);
DispatchEntry entry = DispatchEntry.from(request);
dispatch(entry, writeBatch);
dispatchLMQ(request, writeBatch);
final int msgSize = request.getMsgSize();
final long phyOffset = request.getCommitLogOffset();
if (phyOffset + msgSize >= maxPhyOffset) {
maxPhyOffset = phyOffset + msgSize;
final int msgSize = request.getMsgSize();
final long phyOffset = request.getCommitLogOffset();
if (phyOffset + msgSize >= maxPhyOffset) {
maxPhyOffset = phyOffset + msgSize;
}
}
this.rocksDBConsumeQueueOffsetTable.putMaxPhyAndCqOffset(tempTopicQueueMaxOffsetMap, writeBatch, maxPhyOffset);
this.rocksDBStorage.batchPut(writeBatch);
this.rocksDBConsumeQueueOffsetTable.putHeapMaxCqOffset(tempTopicQueueMaxOffsetMap);
long storeTimeStamp = request.getStoreTimestamp();
long storeTimeStamp = requests.get(size - 1).getStoreTimestamp();
if (this.messageStore.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE
|| this.messageStore.getMessageStoreConfig().isEnableDLegerCommitLog()) {
this.messageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimeStamp);
}
this.messageStore.getStoreCheckpoint().setLogicsMsgTimestamp(storeTimeStamp);
notifyMessageArrival(request);
notifyMessageArriveAndClear(requests);
return true;
} catch (Exception e) {
ERROR_LOG.error("putMessagePosition0 failed.", e);
@@ -311,9 +337,12 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
}
}
private void notifyMessageArrival(DispatchRequest request) {
private void notifyMessageArriveAndClear(List<DispatchRequest> requests) {
try {
this.messageStore.notifyMessageArriveIfNecessary(request);
for (DispatchRequest dp : requests) {
this.messageStore.notifyMessageArriveIfNecessary(dp);
}
requests.clear();
} catch (Exception e) {
ERROR_LOG.error("notifyMessageArriveAndClear Failed.", e);
}
@@ -538,4 +567,8 @@ public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
}
return super.getMaxOffset(topic, queueId);
}
public boolean isStopped() {
return ServiceState.SHUTDOWN_ALREADY == serviceState.get();
}
}
store/src/main/java/org/apache/rocketmq/store/queue/RocksGroupCommitService.java
+103
View File
@@ -0,0 +1,103 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.store.queue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.store.DispatchRequest;
import org.rocksdb.RocksDBException;
public class RocksGroupCommitService extends ServiceThread {
private static final int MAX_BUFFER_SIZE = 100_000;
private static final int PREFERRED_DISPATCH_REQUEST_COUNT = 256;
private final LinkedBlockingQueue<DispatchRequest> buffer;
private final RocksDBConsumeQueueStore store;
private final List<DispatchRequest> requests = new ArrayList<>(PREFERRED_DISPATCH_REQUEST_COUNT);
public RocksGroupCommitService(RocksDBConsumeQueueStore store) {
this.store = store;
this.buffer = new LinkedBlockingQueue<>(MAX_BUFFER_SIZE);
}
@Override
public String getServiceName() {
return "RocksGroupCommit";
}
@Override
public void run() {
log.info("{} service started", this.getServiceName());
while (!this.isStopped()) {
try {
this.waitForRunning(10);
this.doCommit();
} catch (Exception e) {
log.warn("{} service has exception. ", this.getServiceName(), e);
}
}
log.info("{} service end", this.getServiceName());
}
public void putRequest(final DispatchRequest request) throws InterruptedException {
while (!buffer.offer(request, 3, TimeUnit.SECONDS)) {
log.warn("RocksGroupCommitService#buffer is full, 3s elapsed before space becomes available");
}
this.wakeup();
}
private void doCommit() {
while (!buffer.isEmpty()) {
while (true) {
DispatchRequest dispatchRequest = buffer.poll();
if (null != dispatchRequest) {
requests.add(dispatchRequest);
}
if (requests.isEmpty()) {
// buffer has been drained
break;
}
if (null == dispatchRequest || requests.size() >= PREFERRED_DISPATCH_REQUEST_COUNT) {
groupCommit();
}
}
}
}
private void groupCommit() {
while (!store.isStopped()) {
try {
// putMessagePosition will clear requests after consume queue building completion
store.putMessagePosition(requests);
break;
} catch (RocksDBException e) {
log.error("Failed to build consume queue in RocksDB", e);
}
}
}
}
store/src/main/java/org/apache/rocketmq/store/rocksdb/RocksDBOptionsFactory.java
+4 -3
View File
@@ -22,6 +22,7 @@ import org.rocksdb.BlockBasedTableConfig;
import org.rocksdb.BloomFilter;
import org.rocksdb.ColumnFamilyOptions;
import org.rocksdb.CompactionOptionsUniversal;
import org.rocksdb.CompactionPriority;
import org.rocksdb.CompactionStopStyle;
import org.rocksdb.CompactionStyle;
import org.rocksdb.CompressionType;
@@ -79,6 +80,7 @@ public class RocksDBOptionsFactory {
setCompressionType(compressionType).
setBottommostCompressionType(bottomMostCompressionType).
setNumLevels(7).
setCompactionPriority(CompactionPriority.MinOverlappingRatio).
setCompactionStyle(CompactionStyle.UNIVERSAL).
setCompactionOptionsUniversal(compactionOption).
setMaxCompactionBytes(100 * SizeUnit.GB).
@@ -144,10 +146,8 @@ public class RocksDBOptionsFactory {
setInfoLogLevel(InfoLogLevel.INFO_LEVEL).
setWalRecoveryMode(WALRecoveryMode.PointInTimeRecovery).
setManualWalFlush(true).
setMaxTotalWalSize(0).
setWalSizeLimitMB(0).
setWalTtlSeconds(0).
setCreateIfMissing(true).
setBytesPerSync(SizeUnit.MB).
setCreateMissingColumnFamilies(true).
setMaxOpenFiles(-1).
setMaxLogFileSize(SizeUnit.GB).
@@ -156,6 +156,7 @@ public class RocksDBOptionsFactory {
setAllowConcurrentMemtableWrite(false).
setStatistics(statistics).
setAtomicFlush(true).
setCompactionReadaheadSize(4 * SizeUnit.MB).
setMaxBackgroundJobs(32).
setMaxSubcompactions(8).
setParanoidChecks(true).