You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "fujian-zfj (via GitHub)" <gi...@apache.org> on 2023/07/23 03:37:20 UTC
[GitHub] [rocketmq] fujian-zfj opened a new pull request, #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
fujian-zfj opened a new pull request, #7065:
URL: https://github.com/apache/rocketmq/pull/7065
### Which Issue(s) This PR Fixes
Fixes #7064
### Brief Description
This proposal mainly optimizes and solves the existing perfermance problems in the million-topic scenario from two levels:
1. Metadata like topic, subscription, consumerOffset implement kv storage
2. ConsumeQueue index file implements kv storage
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1277069134
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java:
##########
@@ -0,0 +1,1296 @@
+/*
+ * 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.io.File;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.Pair;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.store.DefaultMessageStore;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.FixedSizeCache;
+import org.apache.rocketmq.store.RocksDBConsumeQueue;
+import org.apache.rocketmq.store.config.BrokerRole;
+import org.apache.rocketmq.store.config.StorePathConfigHelper;
+import org.apache.rocketmq.store.rocksdb.ConsumeQueueRocksDBStorage;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.RocksIterator;
+import org.rocksdb.WriteBatch;
+
+public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
+ private static final Logger ERROR_LOG = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);
+ private static final Logger ROCKSDB_LOG = LoggerFactory.getLogger(LoggerName.ROCKSDB_LOGGER_NAME);
+ public static final Charset CHARSET_UTF8 = Charset.forName("UTF-8");
+ private static final byte[] MAX_BYTES = "max".getBytes(CHARSET_UTF8);
+ private static final byte[] MIN_BYTES = "min".getBytes(CHARSET_UTF8);
+ private static final int BATCH_SIZE = 16;
+ private static final byte CTRL_0 = '\u0000';
+ private static final byte CTRL_A = '\u0001';
+ private static final byte CTRL_2 = '\u0002';
+
+ /**
+ * Rocksdb ConsumeQueue's store unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬───────────────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ QueueId │ CTRL_A │ ConsumeQueue Offset │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (4 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴───────────────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────────────┬──────────────────┬──────────────────┬───────────────────────┐
+ * │ CommitLog Physical Offset │ Body Size │ Tag HashCode │ Msg Store Time │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (4 Bytes) │ (8 Bytes) │ (8 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────────────┴──────────────────┴──────────────────┴───────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's store unit. Size:
+ * CommitLog Physical Offset(8) + Body Size(4) + Tag HashCode(8) + Msg Store Time(8) + ConsumeQueue Offset(8) = 36 Bytes
+ */
+ public static final int PHY_OFFSET_OFFSET = 0;
+ public static final int PHY_MSG_LEN_OFFSET = 8;
+ public static final int MSG_TAG_HASHCODE_OFFSET = 12;
+ public static final int MSG_STORE_TIME_SIZE_OFFSET = 20;
+ public static final int CQ_OFFSET_OFFSET = 28;
+ public static final int CQ_UNIT_SIZE = 36;
+
+ private static final int MAX_KEY_LEN = 300;
+ private static final int MAX_VALUE_LEN = CQ_UNIT_SIZE;
+
+ /**
+ * Rocksdb ConsumeQueue's Offset unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬─────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ Max(min) │ CTRL_A │ QueueId │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (3 Bytes) │ (1 Bytes) │ (4 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴─────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────┬────────────────────────┐
+ * │ CommitLog Physical Offset │ CTRL_A │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────┴────────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's Offset unit. Size: CommitLog Physical Offset(8) + CTRL_A(1) + ConsumeQueue Offset(8) = 17 Bytes
+ */
+ private static final int OFFSET_PHY_OFFSET = 0;
+ private static final int OFFSET_CQ_OFFSET = 9;
+
+ private final ScheduledExecutorService scheduledExecutorService;
+
+ private final String storePath;
+ private final ConsumeQueueRocksDBStorage rocksDBStorage;
+ private final FlushOptions flushOptions;
+ private final WriteBatch writeBatch;
+ private final List<DispatchRequest> bufferDRList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> cqBBPairList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> offsetBBPairList;
+
+ private final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap;
+
+ private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinCqOffset;
+ private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ByteBuffer maxPhyOffsetBB;
+
+ private static final ByteBuffer INNER_CHECKPOINT_TOPIC;
+ private static final int INNER_CHECKPOINT_TOPIC_LEN;
+ static {
+ byte[] topicBytes = "CHECKPOINT_TOPIC".getBytes(CHARSET_UTF8);
+ INNER_CHECKPOINT_TOPIC_LEN = 14 + topicBytes.length;
+ INNER_CHECKPOINT_TOPIC = ByteBuffer.allocateDirect(INNER_CHECKPOINT_TOPIC_LEN);
+ buildOffsetKeyBB0(INNER_CHECKPOINT_TOPIC, topicBytes, 0, true);
+ }
+ private volatile boolean isCQError = false;
+
+ private FixedSizeCache<TopicQueueOffset, ByteBuffer> consumeQueueCache;
+
+ public RocksDBConsumeQueueStore(DefaultMessageStore messageStore) {
+ super(messageStore);
+
+ this.storePath = StorePathConfigHelper.getStorePathConsumeQueue(messageStoreConfig.getStorePathRootDir());
+ this.rocksDBStorage = new ConsumeQueueRocksDBStorage(messageStore, storePath, 4);
+ this.flushOptions = new FlushOptions();
+ this.flushOptions.setWaitForFlush(true);
+ this.flushOptions.setAllowWriteStall(false);
+
+ this.writeBatch = new WriteBatch();
+ this.bufferDRList = new ArrayList(BATCH_SIZE);
Review Comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhimins commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "lizhimins (via GitHub)" <gi...@apache.org>.
lizhimins commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1271399698
##########
common/src/main/java/org/apache/rocketmq/common/TopicConfig.java:
##########
@@ -27,7 +29,7 @@
import static org.apache.rocketmq.common.TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE;
-public class TopicConfig {
+public class TopicConfig implements Serializable {
Review Comment:
不建议这么改
##########
common/src/main/java/org/apache/rocketmq/common/TopicConfig.java:
##########
@@ -27,7 +29,7 @@
import static org.apache.rocketmq.common.TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE;
-public class TopicConfig {
+public class TopicConfig implements Serializable {
Review Comment:
不建议这么改
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1275705409
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java:
##########
@@ -0,0 +1,1296 @@
+/*
+ * 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.io.File;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.Pair;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.store.DefaultMessageStore;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.FixedSizeCache;
+import org.apache.rocketmq.store.RocksDBConsumeQueue;
+import org.apache.rocketmq.store.config.BrokerRole;
+import org.apache.rocketmq.store.config.StorePathConfigHelper;
+import org.apache.rocketmq.store.rocksdb.ConsumeQueueRocksDBStorage;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.RocksIterator;
+import org.rocksdb.WriteBatch;
+
+public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
+ private static final Logger ERROR_LOG = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);
+ private static final Logger ROCKSDB_LOG = LoggerFactory.getLogger(LoggerName.ROCKSDB_LOGGER_NAME);
+ public static final Charset CHARSET_UTF8 = Charset.forName("UTF-8");
+ private static final byte[] MAX_BYTES = "max".getBytes(CHARSET_UTF8);
+ private static final byte[] MIN_BYTES = "min".getBytes(CHARSET_UTF8);
+ private static final int BATCH_SIZE = 16;
+ private static final byte CTRL_0 = '\u0000';
+ private static final byte CTRL_A = '\u0001';
+ private static final byte CTRL_2 = '\u0002';
+
+ /**
+ * Rocksdb ConsumeQueue's store unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬───────────────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ QueueId │ CTRL_A │ ConsumeQueue Offset │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (4 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴───────────────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────────────┬──────────────────┬──────────────────┬───────────────────────┐
+ * │ CommitLog Physical Offset │ Body Size │ Tag HashCode │ Msg Store Time │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (4 Bytes) │ (8 Bytes) │ (8 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────────────┴──────────────────┴──────────────────┴───────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's store unit. Size:
+ * CommitLog Physical Offset(8) + Body Size(4) + Tag HashCode(8) + Msg Store Time(8) + ConsumeQueue Offset(8) = 36 Bytes
+ */
+ public static final int PHY_OFFSET_OFFSET = 0;
+ public static final int PHY_MSG_LEN_OFFSET = 8;
+ public static final int MSG_TAG_HASHCODE_OFFSET = 12;
+ public static final int MSG_STORE_TIME_SIZE_OFFSET = 20;
+ public static final int CQ_OFFSET_OFFSET = 28;
+ public static final int CQ_UNIT_SIZE = 36;
+
+ private static final int MAX_KEY_LEN = 300;
+ private static final int MAX_VALUE_LEN = CQ_UNIT_SIZE;
+
+ /**
+ * Rocksdb ConsumeQueue's Offset unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬─────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ Max(min) │ CTRL_A │ QueueId │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (3 Bytes) │ (1 Bytes) │ (4 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴─────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────┬────────────────────────┐
+ * │ CommitLog Physical Offset │ CTRL_A │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────┴────────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's Offset unit. Size: CommitLog Physical Offset(8) + CTRL_A(1) + ConsumeQueue Offset(8) = 17 Bytes
+ */
+ private static final int OFFSET_PHY_OFFSET = 0;
+ private static final int OFFSET_CQ_OFFSET = 9;
+
+ private final ScheduledExecutorService scheduledExecutorService;
+
+ private final String storePath;
+ private final ConsumeQueueRocksDBStorage rocksDBStorage;
+ private final FlushOptions flushOptions;
+ private final WriteBatch writeBatch;
+ private final List<DispatchRequest> bufferDRList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> cqBBPairList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> offsetBBPairList;
+
+ private final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap;
+
+ private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinCqOffset;
+ private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ByteBuffer maxPhyOffsetBB;
+
+ private static final ByteBuffer INNER_CHECKPOINT_TOPIC;
+ private static final int INNER_CHECKPOINT_TOPIC_LEN;
+ static {
+ byte[] topicBytes = "CHECKPOINT_TOPIC".getBytes(CHARSET_UTF8);
+ INNER_CHECKPOINT_TOPIC_LEN = 14 + topicBytes.length;
+ INNER_CHECKPOINT_TOPIC = ByteBuffer.allocateDirect(INNER_CHECKPOINT_TOPIC_LEN);
+ buildOffsetKeyBB0(INNER_CHECKPOINT_TOPIC, topicBytes, 0, true);
+ }
+ private volatile boolean isCQError = false;
+
+ private FixedSizeCache<TopicQueueOffset, ByteBuffer> consumeQueueCache;
+
+ public RocksDBConsumeQueueStore(DefaultMessageStore messageStore) {
+ super(messageStore);
+
+ this.storePath = StorePathConfigHelper.getStorePathConsumeQueue(messageStoreConfig.getStorePathRootDir());
+ this.rocksDBStorage = new ConsumeQueueRocksDBStorage(messageStore, storePath, 4);
+ this.flushOptions = new FlushOptions();
+ this.flushOptions.setWaitForFlush(true);
+ this.flushOptions.setAllowWriteStall(false);
+
+ this.writeBatch = new WriteBatch();
+ this.bufferDRList = new ArrayList(BATCH_SIZE);
Review Comment:
Raw use of parameterized class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] aaron-ai commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "aaron-ai (via GitHub)" <gi...@apache.org>.
aaron-ai commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1271643982
##########
pom.xml:
##########
@@ -973,6 +974,12 @@
<artifactId>jul-to-slf4j</artifactId>
<version>${jul-to-slf4j.version}</version>
</dependency>
+ <!-- rocksdb -->
+ <dependency>
+ <groupId>io.github.aliyunmq</groupId>
+ <artifactId>rocketmq-rocksdb</artifactId>
+ <version>${rocksdb.version}</version>
+ </dependency>
Review Comment:
Is this feature compatible with other platforms (Windows and macOS)? If not, should we set it as optional?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhanhui commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1277120196
##########
broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java:
##########
@@ -225,7 +225,7 @@ public void shutdown() {
public synchronized void changeBrokerRole(final Long newMasterBrokerId, final String newMasterAddress,
final Integer newMasterEpoch,
- final Integer syncStateSetEpoch, final Set<Long> syncStateSet) {
+ final Integer syncStateSetEpoch, final Set<Long> syncStateSet) throws Exception {
Review Comment:
`Exception` is too general...Create one within the project exception hierarchy
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1275712342
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java:
##########
@@ -0,0 +1,1296 @@
+/*
+ * 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.io.File;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.Pair;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.store.DefaultMessageStore;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.FixedSizeCache;
+import org.apache.rocketmq.store.RocksDBConsumeQueue;
+import org.apache.rocketmq.store.config.BrokerRole;
+import org.apache.rocketmq.store.config.StorePathConfigHelper;
+import org.apache.rocketmq.store.rocksdb.ConsumeQueueRocksDBStorage;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.RocksIterator;
+import org.rocksdb.WriteBatch;
+
+public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
+ private static final Logger ERROR_LOG = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);
+ private static final Logger ROCKSDB_LOG = LoggerFactory.getLogger(LoggerName.ROCKSDB_LOGGER_NAME);
+ public static final Charset CHARSET_UTF8 = Charset.forName("UTF-8");
+ private static final byte[] MAX_BYTES = "max".getBytes(CHARSET_UTF8);
+ private static final byte[] MIN_BYTES = "min".getBytes(CHARSET_UTF8);
+ private static final int BATCH_SIZE = 16;
+ private static final byte CTRL_0 = '\u0000';
+ private static final byte CTRL_A = '\u0001';
+ private static final byte CTRL_2 = '\u0002';
+
+ /**
+ * Rocksdb ConsumeQueue's store unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬───────────────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ QueueId │ CTRL_A │ ConsumeQueue Offset │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (4 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴───────────────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────────────┬──────────────────┬──────────────────┬───────────────────────┐
+ * │ CommitLog Physical Offset │ Body Size │ Tag HashCode │ Msg Store Time │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (4 Bytes) │ (8 Bytes) │ (8 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────────────┴──────────────────┴──────────────────┴───────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's store unit. Size:
+ * CommitLog Physical Offset(8) + Body Size(4) + Tag HashCode(8) + Msg Store Time(8) + ConsumeQueue Offset(8) = 36 Bytes
+ */
+ public static final int PHY_OFFSET_OFFSET = 0;
+ public static final int PHY_MSG_LEN_OFFSET = 8;
+ public static final int MSG_TAG_HASHCODE_OFFSET = 12;
+ public static final int MSG_STORE_TIME_SIZE_OFFSET = 20;
+ public static final int CQ_OFFSET_OFFSET = 28;
+ public static final int CQ_UNIT_SIZE = 36;
+
+ private static final int MAX_KEY_LEN = 300;
+ private static final int MAX_VALUE_LEN = CQ_UNIT_SIZE;
+
+ /**
+ * Rocksdb ConsumeQueue's Offset unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬─────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ Max(min) │ CTRL_A │ QueueId │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (3 Bytes) │ (1 Bytes) │ (4 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴─────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────┬────────────────────────┐
+ * │ CommitLog Physical Offset │ CTRL_A │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────┴────────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's Offset unit. Size: CommitLog Physical Offset(8) + CTRL_A(1) + ConsumeQueue Offset(8) = 17 Bytes
+ */
+ private static final int OFFSET_PHY_OFFSET = 0;
+ private static final int OFFSET_CQ_OFFSET = 9;
+
+ private final ScheduledExecutorService scheduledExecutorService;
+
+ private final String storePath;
+ private final ConsumeQueueRocksDBStorage rocksDBStorage;
+ private final FlushOptions flushOptions;
+ private final WriteBatch writeBatch;
+ private final List<DispatchRequest> bufferDRList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> cqBBPairList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> offsetBBPairList;
+
+ private final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap;
+
+ private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinCqOffset;
+ private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ByteBuffer maxPhyOffsetBB;
+
+ private static final ByteBuffer INNER_CHECKPOINT_TOPIC;
+ private static final int INNER_CHECKPOINT_TOPIC_LEN;
+ static {
+ byte[] topicBytes = "CHECKPOINT_TOPIC".getBytes(CHARSET_UTF8);
+ INNER_CHECKPOINT_TOPIC_LEN = 14 + topicBytes.length;
+ INNER_CHECKPOINT_TOPIC = ByteBuffer.allocateDirect(INNER_CHECKPOINT_TOPIC_LEN);
+ buildOffsetKeyBB0(INNER_CHECKPOINT_TOPIC, topicBytes, 0, true);
+ }
+ private volatile boolean isCQError = false;
+
+ private FixedSizeCache<TopicQueueOffset, ByteBuffer> consumeQueueCache;
+
+ public RocksDBConsumeQueueStore(DefaultMessageStore messageStore) {
+ super(messageStore);
+
+ this.storePath = StorePathConfigHelper.getStorePathConsumeQueue(messageStoreConfig.getStorePathRootDir());
+ this.rocksDBStorage = new ConsumeQueueRocksDBStorage(messageStore, storePath, 4);
+ this.flushOptions = new FlushOptions();
+ this.flushOptions.setWaitForFlush(true);
+ this.flushOptions.setAllowWriteStall(false);
+
+ this.writeBatch = new WriteBatch();
+ this.bufferDRList = new ArrayList(BATCH_SIZE);
+ this.cqBBPairList = new ArrayList(BATCH_SIZE);
+ this.offsetBBPairList = new ArrayList(BATCH_SIZE);
+ for (int i = 0; i < BATCH_SIZE; i++) {
+ ByteBuffer bbKey = ByteBuffer.allocateDirect(MAX_KEY_LEN);
+ ByteBuffer bbValue = ByteBuffer.allocateDirect(MAX_VALUE_LEN);
+ this.cqBBPairList.add(new Pair(bbKey, bbValue));
+
+ ByteBuffer offsetKey = ByteBuffer.allocateDirect(MAX_KEY_LEN);
+ ByteBuffer offsetValue = ByteBuffer.allocateDirect(17);
+ this.offsetBBPairList.add(new Pair(offsetKey, offsetValue));
+ }
+ this.tempTopicQueueMaxOffsetMap = new HashMap<>();
+ this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
+
+ this.topicQueueMinCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+
+ this.consumeQueueCache = new FixedSizeCache(1000000);
+
+ this.scheduledExecutorService =
+ Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RocksDBConsumeQueueStoreScheduledThread", messageStore.getBrokerIdentity()));
+ }
+
+ @Override
+ public void start() {
+ log.info("RocksDB ConsumeQueueStore start!");
+ this.scheduledExecutorService.scheduleAtFixedRate(() -> {
+ rocksDBStorage.statRocksdb(ROCKSDB_LOG);
+ }, 10, this.messageStoreConfig.getStatConsumeQueueRocksDbIntervalSec(), TimeUnit.SECONDS);
+
+ this.scheduledExecutorService.scheduleWithFixedDelay(() -> {
+ cleanTopic(messageStore.getTopicConfigs().keySet());
+ }, 10, this.messageStoreConfig.getCleanDirtyConsumeQueueIntervalMin(), TimeUnit.MINUTES);
+ }
+
+ @Override
+ public boolean load() {
+ boolean result = this.rocksDBStorage.start();
+ log.info("load rocksdb consume queue {}.", result ? "OK" : "Failed");
+ return result;
+ }
+
+ @Override
+ public void recover() {
+ // ignored
+ }
+
+ @Override
+ public boolean recoverConcurrently() {
+ return true;
+ }
+
+ @Override
+ public boolean shutdown() {
+ this.scheduledExecutorService.shutdown();
+ return shutdownInner();
+ }
+
+ private boolean shutdownInner() {
+ this.consumeQueueCache.shutdown();
+ return this.rocksDBStorage.shutdown();
+ }
+
+ @Override
+ public void putMessagePositionInfoWrapper(DispatchRequest request) throws Exception {
+ if (request == null || this.bufferDRList.size() >= BATCH_SIZE) {
Review Comment:
Bulk put is used here, if the number of requests never reaches BATCH_SIZE, does it never put?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj closed pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj closed pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
URL: https://github.com/apache/rocketmq/pull/7065
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1271774715
##########
pom.xml:
##########
@@ -973,6 +974,12 @@
<artifactId>jul-to-slf4j</artifactId>
<version>${jul-to-slf4j.version}</version>
</dependency>
+ <!-- rocksdb -->
+ <dependency>
+ <groupId>io.github.aliyunmq</groupId>
+ <artifactId>rocketmq-rocksdb</artifactId>
+ <version>${rocksdb.version}</version>
+ </dependency>
Review Comment:
support Windows, MacOS and Linux
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj commented on pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj commented on PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#issuecomment-1655248837
> Please break this huge PR into smaller ones
OK, I will split it into two PRS, metadata and consumeQueue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhanhui commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1277118325
##########
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java:
##########
@@ -746,7 +760,12 @@ public boolean initializeMetadata() {
public boolean initializeMessageStore() {
boolean result = true;
try {
- DefaultMessageStore defaultMessageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener, this.brokerConfig, topicConfigManager.getTopicConfigTable());
+ DefaultMessageStore defaultMessageStore;
+ if (isEnableRocksDBStore()) {
Review Comment:
Will SPI be a better choice than `if...else`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhanhui commented on pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#issuecomment-1655065609
It's a huge pull request, containing massive changes to core parts of this project with more than **6000** lines of change. Thus, it's challenging to have a quality review. Normally, developers would break it into a series of small and verifiable ones to keep community participation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj commented on pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj commented on PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#issuecomment-1655238959
> Please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage [rocketmq]
Posted by "RongtongJin (via GitHub)" <gi...@apache.org>.
RongtongJin closed pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
URL: https://github.com/apache/rocketmq/pull/7065
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1271406032
##########
common/pom.xml:
##########
@@ -104,5 +104,10 @@
<groupId>io.github.aliyunmq</groupId>
<artifactId>rocketmq-logback-classic</artifactId>
</dependency>
+ <!-- rocksdb -->
+ <dependency>
+ <groupId>io.github.aliyunmq</groupId>
Review Comment:
It's still open source, but we need a new compactionFilter for ConsumeQueue. If you're interested, see how flink uses rocksdb.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] HScarb commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "HScarb (via GitHub)" <gi...@apache.org>.
HScarb commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1275712342
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueStore.java:
##########
@@ -0,0 +1,1296 @@
+/*
+ * 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.io.File;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.Pair;
+import org.apache.rocketmq.common.ThreadFactoryImpl;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.store.DefaultMessageStore;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.FixedSizeCache;
+import org.apache.rocketmq.store.RocksDBConsumeQueue;
+import org.apache.rocketmq.store.config.BrokerRole;
+import org.apache.rocketmq.store.config.StorePathConfigHelper;
+import org.apache.rocketmq.store.rocksdb.ConsumeQueueRocksDBStorage;
+import org.apache.rocketmq.store.timer.TimerMessageStore;
+import org.rocksdb.ColumnFamilyHandle;
+import org.rocksdb.FlushOptions;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.RocksIterator;
+import org.rocksdb.WriteBatch;
+
+public class RocksDBConsumeQueueStore extends AbstractConsumeQueueStore {
+ private static final Logger ERROR_LOG = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);
+ private static final Logger ROCKSDB_LOG = LoggerFactory.getLogger(LoggerName.ROCKSDB_LOGGER_NAME);
+ public static final Charset CHARSET_UTF8 = Charset.forName("UTF-8");
+ private static final byte[] MAX_BYTES = "max".getBytes(CHARSET_UTF8);
+ private static final byte[] MIN_BYTES = "min".getBytes(CHARSET_UTF8);
+ private static final int BATCH_SIZE = 16;
+ private static final byte CTRL_0 = '\u0000';
+ private static final byte CTRL_A = '\u0001';
+ private static final byte CTRL_2 = '\u0002';
+
+ /**
+ * Rocksdb ConsumeQueue's store unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬───────────────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ QueueId │ CTRL_A │ ConsumeQueue Offset │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (4 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴───────────────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────────────┬──────────────────┬──────────────────┬───────────────────────┐
+ * │ CommitLog Physical Offset │ Body Size │ Tag HashCode │ Msg Store Time │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (4 Bytes) │ (8 Bytes) │ (8 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────────────┴──────────────────┴──────────────────┴───────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's store unit. Size:
+ * CommitLog Physical Offset(8) + Body Size(4) + Tag HashCode(8) + Msg Store Time(8) + ConsumeQueue Offset(8) = 36 Bytes
+ */
+ public static final int PHY_OFFSET_OFFSET = 0;
+ public static final int PHY_MSG_LEN_OFFSET = 8;
+ public static final int MSG_TAG_HASHCODE_OFFSET = 12;
+ public static final int MSG_STORE_TIME_SIZE_OFFSET = 20;
+ public static final int CQ_OFFSET_OFFSET = 28;
+ public static final int CQ_UNIT_SIZE = 36;
+
+ private static final int MAX_KEY_LEN = 300;
+ private static final int MAX_VALUE_LEN = CQ_UNIT_SIZE;
+
+ /**
+ * Rocksdb ConsumeQueue's Offset unit. Format:
+ *
+ * <pre>
+ * ┌─────────────────────────┬───────────┬───────────────────────┬───────────┬───────────┬───────────┬─────────────┐
+ * │ Topic Bytes Array Size │ CTRL_A │ Topic Bytes Array │ CTRL_A │ Max(min) │ CTRL_A │ QueueId │
+ * │ (4 Bytes) │ (1 Bytes) │ (n Bytes) │ (1 Bytes) │ (3 Bytes) │ (1 Bytes) │ (4 Bytes) │
+ * ├─────────────────────────┴───────────┴───────────────────────┴───────────┴───────────┴───────────┴─────────────┤
+ * │ Key Unit │
+ * │ │
+ * </pre>
+ *
+ * <pre>
+ * ┌─────────────────────────────┬───────────┬────────────────────────┐
+ * │ CommitLog Physical Offset │ CTRL_A │ ConsumeQueue Offset │
+ * │ (8 Bytes) │ (1 Bytes) │ (8 Bytes) │
+ * ├─────────────────────────────┴───────────┴────────────────────────┤
+ * │ Value Unit │
+ * │ │
+ * </pre>
+ * ConsumeQueue's Offset unit. Size: CommitLog Physical Offset(8) + CTRL_A(1) + ConsumeQueue Offset(8) = 17 Bytes
+ */
+ private static final int OFFSET_PHY_OFFSET = 0;
+ private static final int OFFSET_CQ_OFFSET = 9;
+
+ private final ScheduledExecutorService scheduledExecutorService;
+
+ private final String storePath;
+ private final ConsumeQueueRocksDBStorage rocksDBStorage;
+ private final FlushOptions flushOptions;
+ private final WriteBatch writeBatch;
+ private final List<DispatchRequest> bufferDRList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> cqBBPairList;
+ private final List<Pair<ByteBuffer, ByteBuffer>> offsetBBPairList;
+
+ private final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap;
+
+ private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinCqOffset;
+ private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ByteBuffer maxPhyOffsetBB;
+
+ private static final ByteBuffer INNER_CHECKPOINT_TOPIC;
+ private static final int INNER_CHECKPOINT_TOPIC_LEN;
+ static {
+ byte[] topicBytes = "CHECKPOINT_TOPIC".getBytes(CHARSET_UTF8);
+ INNER_CHECKPOINT_TOPIC_LEN = 14 + topicBytes.length;
+ INNER_CHECKPOINT_TOPIC = ByteBuffer.allocateDirect(INNER_CHECKPOINT_TOPIC_LEN);
+ buildOffsetKeyBB0(INNER_CHECKPOINT_TOPIC, topicBytes, 0, true);
+ }
+ private volatile boolean isCQError = false;
+
+ private FixedSizeCache<TopicQueueOffset, ByteBuffer> consumeQueueCache;
+
+ public RocksDBConsumeQueueStore(DefaultMessageStore messageStore) {
+ super(messageStore);
+
+ this.storePath = StorePathConfigHelper.getStorePathConsumeQueue(messageStoreConfig.getStorePathRootDir());
+ this.rocksDBStorage = new ConsumeQueueRocksDBStorage(messageStore, storePath, 4);
+ this.flushOptions = new FlushOptions();
+ this.flushOptions.setWaitForFlush(true);
+ this.flushOptions.setAllowWriteStall(false);
+
+ this.writeBatch = new WriteBatch();
+ this.bufferDRList = new ArrayList(BATCH_SIZE);
+ this.cqBBPairList = new ArrayList(BATCH_SIZE);
+ this.offsetBBPairList = new ArrayList(BATCH_SIZE);
+ for (int i = 0; i < BATCH_SIZE; i++) {
+ ByteBuffer bbKey = ByteBuffer.allocateDirect(MAX_KEY_LEN);
+ ByteBuffer bbValue = ByteBuffer.allocateDirect(MAX_VALUE_LEN);
+ this.cqBBPairList.add(new Pair(bbKey, bbValue));
+
+ ByteBuffer offsetKey = ByteBuffer.allocateDirect(MAX_KEY_LEN);
+ ByteBuffer offsetValue = ByteBuffer.allocateDirect(17);
+ this.offsetBBPairList.add(new Pair(offsetKey, offsetValue));
+ }
+ this.tempTopicQueueMaxOffsetMap = new HashMap<>();
+ this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
+
+ this.topicQueueMinCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+
+ this.consumeQueueCache = new FixedSizeCache(1000000);
+
+ this.scheduledExecutorService =
+ Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("RocksDBConsumeQueueStoreScheduledThread", messageStore.getBrokerIdentity()));
+ }
+
+ @Override
+ public void start() {
+ log.info("RocksDB ConsumeQueueStore start!");
+ this.scheduledExecutorService.scheduleAtFixedRate(() -> {
+ rocksDBStorage.statRocksdb(ROCKSDB_LOG);
+ }, 10, this.messageStoreConfig.getStatConsumeQueueRocksDbIntervalSec(), TimeUnit.SECONDS);
+
+ this.scheduledExecutorService.scheduleWithFixedDelay(() -> {
+ cleanTopic(messageStore.getTopicConfigs().keySet());
+ }, 10, this.messageStoreConfig.getCleanDirtyConsumeQueueIntervalMin(), TimeUnit.MINUTES);
+ }
+
+ @Override
+ public boolean load() {
+ boolean result = this.rocksDBStorage.start();
+ log.info("load rocksdb consume queue {}.", result ? "OK" : "Failed");
+ return result;
+ }
+
+ @Override
+ public void recover() {
+ // ignored
+ }
+
+ @Override
+ public boolean recoverConcurrently() {
+ return true;
+ }
+
+ @Override
+ public boolean shutdown() {
+ this.scheduledExecutorService.shutdown();
+ return shutdownInner();
+ }
+
+ private boolean shutdownInner() {
+ this.consumeQueueCache.shutdown();
+ return this.rocksDBStorage.shutdown();
+ }
+
+ @Override
+ public void putMessagePositionInfoWrapper(DispatchRequest request) throws Exception {
+ if (request == null || this.bufferDRList.size() >= BATCH_SIZE) {
Review Comment:
Bulk put is used here, if the number of requests never reaches BATCH_SIZE, does it never put?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] codecov-commenter commented on pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#issuecomment-1649392395
## [Codecov](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#7065](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (49505a1) into [develop](https://app.codecov.io/gh/apache/rocketmq/commit/90c5382aee07879a80309f257f04114201ccaac6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (90c5382) will **decrease** coverage by `0.08%`.
> The diff coverage is `40.57%`.
```diff
@@ Coverage Diff @@
## develop #7065 +/- ##
=============================================
- Coverage 42.72% 42.65% -0.08%
- Complexity 9282 9477 +195
=============================================
Files 1138 1157 +19
Lines 81146 83192 +2046
Branches 10619 10850 +231
=============================================
+ Hits 34672 35483 +811
- Misses 42137 43292 +1155
- Partials 4337 4417 +80
```
| [Files Changed](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...he/rocketmq/broker/controller/ReplicasManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvY29udHJvbGxlci9SZXBsaWNhc01hbmFnZXIuamF2YQ==) | `46.27% <ø> (ø)` | |
| [...mq/broker/offset/RocksDBConsumerOffsetManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvb2Zmc2V0L1JvY2tzREJDb25zdW1lck9mZnNldE1hbmFnZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...ocketmq/broker/processor/SendMessageProcessor.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvcHJvY2Vzc29yL1NlbmRNZXNzYWdlUHJvY2Vzc29yLmphdmE=) | `36.29% <ø> (ø)` | |
| [.../subscription/RocksDBLmqConsumerOffsetManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvc3Vic2NyaXB0aW9uL1JvY2tzREJMbXFDb25zdW1lck9mZnNldE1hbmFnZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...bscription/RocksDBLmqSubscriptionGroupManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvc3Vic2NyaXB0aW9uL1JvY2tzREJMbXFTdWJzY3JpcHRpb25Hcm91cE1hbmFnZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [.../subscription/RocksDBSubscriptionGroupManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvc3Vic2NyaXB0aW9uL1JvY2tzREJTdWJzY3JpcHRpb25Hcm91cE1hbmFnZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...tmq/broker/topic/RocksDBLmqTopicConfigManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvdG9waWMvUm9ja3NEQkxtcVRvcGljQ29uZmlnTWFuYWdlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...cketmq/broker/topic/RocksDBTopicConfigManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvdG9waWMvUm9ja3NEQlRvcGljQ29uZmlnTWFuYWdlci5qYXZh) | `0.00% <0.00%> (ø)` | |
| [.../consumer/store/RocksDBOffsetSerializeWrapper.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y2xpZW50L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jbGllbnQvY29uc3VtZXIvc3RvcmUvUm9ja3NEQk9mZnNldFNlcmlhbGl6ZVdyYXBwZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...java/org/apache/rocketmq/common/ConfigManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9jb21tb24vQ29uZmlnTWFuYWdlci5qYXZh) | `56.41% <0.00%> (-3.05%)` | :arrow_down: |
| ... and [41 more](https://app.codecov.io/gh/apache/rocketmq/pull/7065?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
... and [25 files with indirect coverage changes](https://app.codecov.io/gh/apache/rocketmq/pull/7065/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1277153179
##########
broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java:
##########
@@ -235,7 +235,7 @@ public synchronized void changeBrokerRole(final Long newMasterBrokerId, final St
}
}
- public void changeToMaster(final int newMasterEpoch, final int syncStateSetEpoch, final Set<Long> syncStateSet) {
+ public void changeToMaster(final int newMasterEpoch, final int syncStateSetEpoch, final Set<Long> syncStateSet) throws Exception {
Review Comment:
OK
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhanhui commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1277120319
##########
broker/src/main/java/org/apache/rocketmq/broker/controller/ReplicasManager.java:
##########
@@ -235,7 +235,7 @@ public synchronized void changeBrokerRole(final Long newMasterBrokerId, final St
}
}
- public void changeToMaster(final int newMasterEpoch, final int syncStateSetEpoch, final Set<Long> syncStateSet) {
+ public void changeToMaster(final int newMasterEpoch, final int syncStateSetEpoch, final Set<Long> syncStateSet) throws Exception {
Review Comment:
Same here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhimins commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "lizhimins (via GitHub)" <gi...@apache.org>.
lizhimins commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1271400344
##########
common/pom.xml:
##########
@@ -104,5 +104,10 @@
<groupId>io.github.aliyunmq</groupId>
<artifactId>rocketmq-logback-classic</artifactId>
</dependency>
+ <!-- rocksdb -->
+ <dependency>
+ <groupId>io.github.aliyunmq</groupId>
Review Comment:
Why rely on a commercial version? How to ensure the stability of this version?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] lizhanhui commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1277150541
##########
common/pom.xml:
##########
@@ -104,5 +104,10 @@
<groupId>io.github.aliyunmq</groupId>
<artifactId>rocketmq-logback-classic</artifactId>
</dependency>
+ <!-- rocksdb -->
+ <dependency>
+ <groupId>io.github.aliyunmq</groupId>
Review Comment:
@fujian-zfj Your purpose of building custom rocksdb is to define a compaction filter. What a custom compaction filter expects is native function. Will this way be better alternative:
1. Use official rocksdb-java jar;
2. Build a shared library(dynamically linked with rocksdb shared library) for expected custom compaction filters;
3. When configuring RocksDB, pass the native function to it, because it does not matter which shared library the compaction filter stays in;
As a matter of fact, this is how the native programming languages use rocksdb in case of dynamic link.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq] fujian-zfj commented on a diff in pull request #7065: [ISSUE #7064] [RIP-66] Support KV(RocksDB) Storage
Posted by "fujian-zfj (via GitHub)" <gi...@apache.org>.
fujian-zfj commented on code in PR #7065:
URL: https://github.com/apache/rocketmq/pull/7065#discussion_r1271807019
##########
common/src/main/java/org/apache/rocketmq/common/TopicConfig.java:
##########
@@ -27,7 +29,7 @@
import static org.apache.rocketmq.common.TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE;
-public class TopicConfig {
+public class TopicConfig implements Serializable {
Review Comment:
已删除
##########
common/src/main/java/org/apache/rocketmq/common/TopicConfig.java:
##########
@@ -27,7 +29,7 @@
import static org.apache.rocketmq.common.TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE;
-public class TopicConfig {
+public class TopicConfig implements Serializable {
Review Comment:
已删除
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org