You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "lizhimins (via GitHub)" <gi...@apache.org> on 2024/03/11 13:02:04 UTC
[PR] [ISSUE #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
lizhimins opened a new pull request, #7899:
URL: https://github.com/apache/rocketmq/pull/7899
<!-- Please make sure the target branch is right. In most case, the target branch should be `develop`. -->
### Which Issue(s) This PR Fixes
<!-- Please ensure that the related issue has already been created, and [link this pull request to that issue using keywords](<https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword>) to ensure automatic closure. -->
Fixes #7878
### Brief Description
<!-- Write a brief description for your pull request to help the maintainer understand the reasons behind your changes. -->
### How Did You Test This Change?
<!-- In order to ensure the code quality of Apache RocketMQ, we expect every pull request to have undergone thorough testing. -->
--
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 #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899#issuecomment-1988441828
## [Codecov](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `84.13534%` with `211 lines` in your changes are missing coverage. Please review.
> Project coverage is 43.09%. Comparing base [(`db4c7da`)](https://app.codecov.io/gh/apache/rocketmq/commit/db4c7da4f7b3792db7c491b0d1937fbaf2670181?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`c71f5b7`)](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
| [Files](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...etmq/tieredstore/core/MessageStoreFetcherImpl.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL2NvcmUvTWVzc2FnZVN0b3JlRmV0Y2hlckltcGwuamF2YQ==) | 77.92% | [39 Missing and 12 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...q/tieredstore/core/MessageStoreDispatcherImpl.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL2NvcmUvTWVzc2FnZVN0b3JlRGlzcGF0Y2hlckltcGwuamF2YQ==) | 69.86% | [25 Missing and 19 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/rocketmq/tieredstore/TieredMessageStore.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL1RpZXJlZE1lc3NhZ2VTdG9yZS5qYXZh) | 70.78% | [9 Missing and 17 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...che/rocketmq/tieredstore/file/FlatMessageFile.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL2ZpbGUvRmxhdE1lc3NhZ2VGaWxlLmphdmE=) | 89.61% | [7 Missing and 9 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../rocketmq/tieredstore/index/IndexStoreService.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL2luZGV4L0luZGV4U3RvcmVTZXJ2aWNlLmphdmE=) | 72.54% | [7 Missing and 7 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...tieredstore/metrics/TieredStoreMetricsManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL21ldHJpY3MvVGllcmVkU3RvcmVNZXRyaWNzTWFuYWdlci5qYXZh) | 18.75% | [13 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ache/rocketmq/tieredstore/file/FlatAppendFile.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL2ZpbGUvRmxhdEFwcGVuZEZpbGUuamF2YQ==) | 91.05% | [5 Missing and 6 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/rocketmq/tieredstore/file/FlatFileStore.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL2ZpbGUvRmxhdEZpbGVTdG9yZS5qYXZh) | 85.52% | [9 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ocketmq/tieredstore/provider/PosixFileSegment.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL3Byb3ZpZGVyL1Bvc2l4RmlsZVNlZ21lbnQuamF2YQ==) | 84.90% | [4 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/rocketmq/tieredstore/MessageStoreConfig.java](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL01lc3NhZ2VTdG9yZUNvbmZpZy5qYXZh) | 88.88% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [8 more](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## develop #7899 +/- ##
=============================================
+ Coverage 42.92% 43.09% +0.17%
+ Complexity 9909 9865 -44
=============================================
Files 1190 1184 -6
Lines 86042 85176 -866
Branches 11081 10977 -104
=============================================
- Hits 36935 36710 -225
+ Misses 44531 43987 -544
+ Partials 4576 4479 -97
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/rocketmq/pull/7899?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?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
Re: [PR] [ISSUE #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "lizhimins (via GitHub)" <gi...@apache.org>.
lizhimins commented on code in PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899#discussion_r1525830303
##########
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.tieredstore.core;
+
+import io.opentelemetry.api.common.Attributes;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
+import org.apache.rocketmq.store.queue.CqUnit;
+import org.apache.rocketmq.tieredstore.MessageStoreConfig;
+import org.apache.rocketmq.tieredstore.MessageStoreExecutor;
+import org.apache.rocketmq.tieredstore.TieredMessageStore;
+import org.apache.rocketmq.tieredstore.common.AppendResult;
+import org.apache.rocketmq.tieredstore.common.FileSegmentType;
+import org.apache.rocketmq.tieredstore.file.FlatFileInterface;
+import org.apache.rocketmq.tieredstore.file.FlatFileStore;
+import org.apache.rocketmq.tieredstore.index.IndexService;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
+import org.apache.rocketmq.tieredstore.util.MessageFormatUtil;
+import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MessageStoreDispatcherImpl extends ServiceThread implements MessageStoreDispatcher {
+
+ protected static final Logger log = LoggerFactory.getLogger(MessageStoreUtil.TIERED_STORE_LOGGER_NAME);
+
+ protected final String brokerName;
+ protected final MessageStore defaultStore;
+ protected final MessageStoreConfig storeConfig;
+ protected final TieredMessageStore messageStore;
+ protected final FlatFileStore flatFileStore;
+ protected final MessageStoreExecutor storeExecutor;
+ protected final MessageStoreFilter topicFilter;
+ protected final Semaphore semaphore;
+ protected final IndexService indexService;
+
+ public MessageStoreDispatcherImpl(TieredMessageStore messageStore) {
+ this.messageStore = messageStore;
+ this.storeConfig = messageStore.getStoreConfig();
+ this.defaultStore = messageStore.getDefaultStore();
+ this.brokerName = storeConfig.getBrokerName();
+ this.semaphore = new Semaphore(
+ this.storeConfig.getTieredStoreMaxPendingLimit() / 4);
+ this.topicFilter = messageStore.getTopicFilter();
+ this.flatFileStore = messageStore.getFlatFileStore();
+ this.storeExecutor = messageStore.getStoreExecutor();
+ this.indexService = messageStore.getIndexService();
+ }
+
+ @Override
+ public String getServiceName() {
+ return MessageStoreDispatcher.class.getSimpleName();
+ }
+
+ public void dispatchWithSemaphore(FlatFileInterface flatFile) {
+ try {
+ if (stopped) {
+ return;
+ }
+ semaphore.acquire();
+ this.dispatchAsync(flatFile, false)
+ .whenComplete((future, throwable) -> semaphore.release());
+ } catch (InterruptedException e) {
+ semaphore.release();
+ }
+ }
+
+ @Override
+ public void dispatch(DispatchRequest request) {
+ if (stopped || topicFilter != null && topicFilter.filterTopic(request.getTopic())) {
+ return;
+ }
+ flatFileStore.computeIfAbsent(
+ new MessageQueue(request.getTopic(), brokerName, request.getQueueId()));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> dispatchAsync(FlatFileInterface flatFile, boolean force) {
+ if (stopped) {
+ return CompletableFuture.completedFuture(true);
+ }
+
+ String topic = flatFile.getMessageQueue().getTopic();
+ int queueId = flatFile.getMessageQueue().getQueueId();
+
+ force = !storeConfig.isTieredStoreGroupCommit() || force;
+ if (force) {
+ flatFile.getFileLock().lock();
+ } else {
+ if (!flatFile.getFileLock().tryLock()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ }
+
+ try {
+ if (topicFilter != null && topicFilter.filterTopic(flatFile.getMessageQueue().getTopic())) {
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long currentOffset = flatFile.getConsumeQueueMaxOffset();
+ long commitOffset = flatFile.getConsumeQueueCommitOffset();
+ long minOffsetInQueue = defaultStore.getMinOffsetInQueue(topic, queueId);
+ long maxOffsetInQueue = defaultStore.getMaxOffsetInQueue(topic, queueId);
+
+ // If set to max offset here, some written messages may be lost
+ if (!flatFile.isFlatFileInit()) {
+ currentOffset = Math.max(minOffsetInQueue,
+ maxOffsetInQueue - storeConfig.getTieredStoreGroupCommitSize());
+ flatFile.initOffset(currentOffset);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ // If last commit failed
+ if (commitOffset < currentOffset) {
+ this.commitAsync(flatFile);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset == maxOffsetInQueue) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset < minOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too small, " +
+ "topic={}, queueId={}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ flatFileStore.computeIfAbsent(new MessageQueue(topic, brokerName, queueId));
+ return CompletableFuture.completedFuture(true);
+ }
+
+ if (currentOffset > maxOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too large, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long interval = TimeUnit.HOURS.toMillis(storeConfig.getCommitLogRollingInterval());
+ if (flatFile.rollingFile(interval)) {
+ log.warn("MessageDispatcher#dispatch, rolling file, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ }
+
+ long bufferSize = 0L;
+ long groupCommitSize = storeConfig.getTieredStoreGroupCommitSize();
+ long groupCommitCount = storeConfig.getTieredStoreGroupCommitCount();
+ long targetOffset = Math.min(currentOffset + groupCommitCount, maxOffsetInQueue);
+
+ ConsumeQueueInterface consumeQueue = defaultStore.getConsumeQueue(topic, queueId);
+ CqUnit cqUnit = consumeQueue.get(currentOffset);
+ SelectMappedBufferResult message =
+ defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
+ boolean timeout = MessageFormatUtil.getStoreTimeStamp(message.getByteBuffer()) +
+ TimeUnit.SECONDS.toMillis(30) < System.currentTimeMillis();
Review Comment:
把这个改成配置了，之前的设计里，延迟取决于定时任务的时间，全部数据只要一来就会缓存在内存里，内存压力高。这里不加这个超时条件也是可以的，会弱化攒批的效果。攒批可以减少对 dfs 的调用次数。
--
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 #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "bxfjb (via GitHub)" <gi...@apache.org>.
bxfjb commented on code in PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899#discussion_r1525734250
##########
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.tieredstore.core;
+
+import io.opentelemetry.api.common.Attributes;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
+import org.apache.rocketmq.store.queue.CqUnit;
+import org.apache.rocketmq.tieredstore.MessageStoreConfig;
+import org.apache.rocketmq.tieredstore.MessageStoreExecutor;
+import org.apache.rocketmq.tieredstore.TieredMessageStore;
+import org.apache.rocketmq.tieredstore.common.AppendResult;
+import org.apache.rocketmq.tieredstore.common.FileSegmentType;
+import org.apache.rocketmq.tieredstore.file.FlatFileInterface;
+import org.apache.rocketmq.tieredstore.file.FlatFileStore;
+import org.apache.rocketmq.tieredstore.index.IndexService;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
+import org.apache.rocketmq.tieredstore.util.MessageFormatUtil;
+import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MessageStoreDispatcherImpl extends ServiceThread implements MessageStoreDispatcher {
+
+ protected static final Logger log = LoggerFactory.getLogger(MessageStoreUtil.TIERED_STORE_LOGGER_NAME);
+
+ protected final String brokerName;
+ protected final MessageStore defaultStore;
+ protected final MessageStoreConfig storeConfig;
+ protected final TieredMessageStore messageStore;
+ protected final FlatFileStore flatFileStore;
+ protected final MessageStoreExecutor storeExecutor;
+ protected final MessageStoreFilter topicFilter;
+ protected final Semaphore semaphore;
+ protected final IndexService indexService;
+
+ public MessageStoreDispatcherImpl(TieredMessageStore messageStore) {
+ this.messageStore = messageStore;
+ this.storeConfig = messageStore.getStoreConfig();
+ this.defaultStore = messageStore.getDefaultStore();
+ this.brokerName = storeConfig.getBrokerName();
+ this.semaphore = new Semaphore(
+ this.storeConfig.getTieredStoreMaxPendingLimit() / 4);
+ this.topicFilter = messageStore.getTopicFilter();
+ this.flatFileStore = messageStore.getFlatFileStore();
+ this.storeExecutor = messageStore.getStoreExecutor();
+ this.indexService = messageStore.getIndexService();
+ }
+
+ @Override
+ public String getServiceName() {
+ return MessageStoreDispatcher.class.getSimpleName();
+ }
+
+ public void dispatchWithSemaphore(FlatFileInterface flatFile) {
+ try {
+ if (stopped) {
+ return;
+ }
+ semaphore.acquire();
+ this.dispatchAsync(flatFile, false)
+ .whenComplete((future, throwable) -> semaphore.release());
+ } catch (InterruptedException e) {
+ semaphore.release();
+ }
+ }
+
+ @Override
+ public void dispatch(DispatchRequest request) {
+ if (stopped || topicFilter != null && topicFilter.filterTopic(request.getTopic())) {
+ return;
+ }
+ flatFileStore.computeIfAbsent(
+ new MessageQueue(request.getTopic(), brokerName, request.getQueueId()));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> dispatchAsync(FlatFileInterface flatFile, boolean force) {
+ if (stopped) {
+ return CompletableFuture.completedFuture(true);
+ }
+
+ String topic = flatFile.getMessageQueue().getTopic();
+ int queueId = flatFile.getMessageQueue().getQueueId();
+
+ force = !storeConfig.isTieredStoreGroupCommit() || force;
+ if (force) {
+ flatFile.getFileLock().lock();
+ } else {
+ if (!flatFile.getFileLock().tryLock()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ }
+
+ try {
+ if (topicFilter != null && topicFilter.filterTopic(flatFile.getMessageQueue().getTopic())) {
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long currentOffset = flatFile.getConsumeQueueMaxOffset();
+ long commitOffset = flatFile.getConsumeQueueCommitOffset();
+ long minOffsetInQueue = defaultStore.getMinOffsetInQueue(topic, queueId);
+ long maxOffsetInQueue = defaultStore.getMaxOffsetInQueue(topic, queueId);
+
+ // If set to max offset here, some written messages may be lost
+ if (!flatFile.isFlatFileInit()) {
+ currentOffset = Math.max(minOffsetInQueue,
+ maxOffsetInQueue - storeConfig.getTieredStoreGroupCommitSize());
+ flatFile.initOffset(currentOffset);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ // If last commit failed
+ if (commitOffset < currentOffset) {
+ this.commitAsync(flatFile);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset == maxOffsetInQueue) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset < minOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too small, " +
+ "topic={}, queueId={}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ flatFileStore.computeIfAbsent(new MessageQueue(topic, brokerName, queueId));
+ return CompletableFuture.completedFuture(true);
+ }
+
+ if (currentOffset > maxOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too large, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long interval = TimeUnit.HOURS.toMillis(storeConfig.getCommitLogRollingInterval());
+ if (flatFile.rollingFile(interval)) {
+ log.warn("MessageDispatcher#dispatch, rolling file, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ }
+
+ long bufferSize = 0L;
+ long groupCommitSize = storeConfig.getTieredStoreGroupCommitSize();
+ long groupCommitCount = storeConfig.getTieredStoreGroupCommitCount();
+ long targetOffset = Math.min(currentOffset + groupCommitCount, maxOffsetInQueue);
+
+ ConsumeQueueInterface consumeQueue = defaultStore.getConsumeQueue(topic, queueId);
+ CqUnit cqUnit = consumeQueue.get(currentOffset);
+ SelectMappedBufferResult message =
+ defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
+ boolean timeout = MessageFormatUtil.getStoreTimeStamp(message.getByteBuffer()) +
+ TimeUnit.SECONDS.toMillis(30) < System.currentTimeMillis();
Review Comment:
你好想问下这里为什么设置了 30s 的超时时间再转储，之前似乎是没有这个逻辑的，定时任务只要发现有数据就直接调用 flatFile.append 了
--
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 #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "lizhimins (via GitHub)" <gi...@apache.org>.
lizhimins commented on code in PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899#discussion_r1525830303
##########
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.tieredstore.core;
+
+import io.opentelemetry.api.common.Attributes;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
+import org.apache.rocketmq.store.queue.CqUnit;
+import org.apache.rocketmq.tieredstore.MessageStoreConfig;
+import org.apache.rocketmq.tieredstore.MessageStoreExecutor;
+import org.apache.rocketmq.tieredstore.TieredMessageStore;
+import org.apache.rocketmq.tieredstore.common.AppendResult;
+import org.apache.rocketmq.tieredstore.common.FileSegmentType;
+import org.apache.rocketmq.tieredstore.file.FlatFileInterface;
+import org.apache.rocketmq.tieredstore.file.FlatFileStore;
+import org.apache.rocketmq.tieredstore.index.IndexService;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
+import org.apache.rocketmq.tieredstore.util.MessageFormatUtil;
+import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MessageStoreDispatcherImpl extends ServiceThread implements MessageStoreDispatcher {
+
+ protected static final Logger log = LoggerFactory.getLogger(MessageStoreUtil.TIERED_STORE_LOGGER_NAME);
+
+ protected final String brokerName;
+ protected final MessageStore defaultStore;
+ protected final MessageStoreConfig storeConfig;
+ protected final TieredMessageStore messageStore;
+ protected final FlatFileStore flatFileStore;
+ protected final MessageStoreExecutor storeExecutor;
+ protected final MessageStoreFilter topicFilter;
+ protected final Semaphore semaphore;
+ protected final IndexService indexService;
+
+ public MessageStoreDispatcherImpl(TieredMessageStore messageStore) {
+ this.messageStore = messageStore;
+ this.storeConfig = messageStore.getStoreConfig();
+ this.defaultStore = messageStore.getDefaultStore();
+ this.brokerName = storeConfig.getBrokerName();
+ this.semaphore = new Semaphore(
+ this.storeConfig.getTieredStoreMaxPendingLimit() / 4);
+ this.topicFilter = messageStore.getTopicFilter();
+ this.flatFileStore = messageStore.getFlatFileStore();
+ this.storeExecutor = messageStore.getStoreExecutor();
+ this.indexService = messageStore.getIndexService();
+ }
+
+ @Override
+ public String getServiceName() {
+ return MessageStoreDispatcher.class.getSimpleName();
+ }
+
+ public void dispatchWithSemaphore(FlatFileInterface flatFile) {
+ try {
+ if (stopped) {
+ return;
+ }
+ semaphore.acquire();
+ this.dispatchAsync(flatFile, false)
+ .whenComplete((future, throwable) -> semaphore.release());
+ } catch (InterruptedException e) {
+ semaphore.release();
+ }
+ }
+
+ @Override
+ public void dispatch(DispatchRequest request) {
+ if (stopped || topicFilter != null && topicFilter.filterTopic(request.getTopic())) {
+ return;
+ }
+ flatFileStore.computeIfAbsent(
+ new MessageQueue(request.getTopic(), brokerName, request.getQueueId()));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> dispatchAsync(FlatFileInterface flatFile, boolean force) {
+ if (stopped) {
+ return CompletableFuture.completedFuture(true);
+ }
+
+ String topic = flatFile.getMessageQueue().getTopic();
+ int queueId = flatFile.getMessageQueue().getQueueId();
+
+ force = !storeConfig.isTieredStoreGroupCommit() || force;
+ if (force) {
+ flatFile.getFileLock().lock();
+ } else {
+ if (!flatFile.getFileLock().tryLock()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ }
+
+ try {
+ if (topicFilter != null && topicFilter.filterTopic(flatFile.getMessageQueue().getTopic())) {
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long currentOffset = flatFile.getConsumeQueueMaxOffset();
+ long commitOffset = flatFile.getConsumeQueueCommitOffset();
+ long minOffsetInQueue = defaultStore.getMinOffsetInQueue(topic, queueId);
+ long maxOffsetInQueue = defaultStore.getMaxOffsetInQueue(topic, queueId);
+
+ // If set to max offset here, some written messages may be lost
+ if (!flatFile.isFlatFileInit()) {
+ currentOffset = Math.max(minOffsetInQueue,
+ maxOffsetInQueue - storeConfig.getTieredStoreGroupCommitSize());
+ flatFile.initOffset(currentOffset);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ // If last commit failed
+ if (commitOffset < currentOffset) {
+ this.commitAsync(flatFile);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset == maxOffsetInQueue) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset < minOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too small, " +
+ "topic={}, queueId={}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ flatFileStore.computeIfAbsent(new MessageQueue(topic, brokerName, queueId));
+ return CompletableFuture.completedFuture(true);
+ }
+
+ if (currentOffset > maxOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too large, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long interval = TimeUnit.HOURS.toMillis(storeConfig.getCommitLogRollingInterval());
+ if (flatFile.rollingFile(interval)) {
+ log.warn("MessageDispatcher#dispatch, rolling file, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ }
+
+ long bufferSize = 0L;
+ long groupCommitSize = storeConfig.getTieredStoreGroupCommitSize();
+ long groupCommitCount = storeConfig.getTieredStoreGroupCommitCount();
+ long targetOffset = Math.min(currentOffset + groupCommitCount, maxOffsetInQueue);
+
+ ConsumeQueueInterface consumeQueue = defaultStore.getConsumeQueue(topic, queueId);
+ CqUnit cqUnit = consumeQueue.get(currentOffset);
+ SelectMappedBufferResult message =
+ defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
+ boolean timeout = MessageFormatUtil.getStoreTimeStamp(message.getByteBuffer()) +
+ TimeUnit.SECONDS.toMillis(30) < System.currentTimeMillis();
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
Re: [PR] [ISSUE #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "bxfjb (via GitHub)" <gi...@apache.org>.
bxfjb commented on code in PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899#discussion_r1525862549
##########
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.tieredstore.core;
+
+import io.opentelemetry.api.common.Attributes;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
+import org.apache.rocketmq.store.queue.CqUnit;
+import org.apache.rocketmq.tieredstore.MessageStoreConfig;
+import org.apache.rocketmq.tieredstore.MessageStoreExecutor;
+import org.apache.rocketmq.tieredstore.TieredMessageStore;
+import org.apache.rocketmq.tieredstore.common.AppendResult;
+import org.apache.rocketmq.tieredstore.common.FileSegmentType;
+import org.apache.rocketmq.tieredstore.file.FlatFileInterface;
+import org.apache.rocketmq.tieredstore.file.FlatFileStore;
+import org.apache.rocketmq.tieredstore.index.IndexService;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
+import org.apache.rocketmq.tieredstore.util.MessageFormatUtil;
+import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MessageStoreDispatcherImpl extends ServiceThread implements MessageStoreDispatcher {
+
+ protected static final Logger log = LoggerFactory.getLogger(MessageStoreUtil.TIERED_STORE_LOGGER_NAME);
+
+ protected final String brokerName;
+ protected final MessageStore defaultStore;
+ protected final MessageStoreConfig storeConfig;
+ protected final TieredMessageStore messageStore;
+ protected final FlatFileStore flatFileStore;
+ protected final MessageStoreExecutor storeExecutor;
+ protected final MessageStoreFilter topicFilter;
+ protected final Semaphore semaphore;
+ protected final IndexService indexService;
+
+ public MessageStoreDispatcherImpl(TieredMessageStore messageStore) {
+ this.messageStore = messageStore;
+ this.storeConfig = messageStore.getStoreConfig();
+ this.defaultStore = messageStore.getDefaultStore();
+ this.brokerName = storeConfig.getBrokerName();
+ this.semaphore = new Semaphore(
+ this.storeConfig.getTieredStoreMaxPendingLimit() / 4);
+ this.topicFilter = messageStore.getTopicFilter();
+ this.flatFileStore = messageStore.getFlatFileStore();
+ this.storeExecutor = messageStore.getStoreExecutor();
+ this.indexService = messageStore.getIndexService();
+ }
+
+ @Override
+ public String getServiceName() {
+ return MessageStoreDispatcher.class.getSimpleName();
+ }
+
+ public void dispatchWithSemaphore(FlatFileInterface flatFile) {
+ try {
+ if (stopped) {
+ return;
+ }
+ semaphore.acquire();
+ this.dispatchAsync(flatFile, false)
+ .whenComplete((future, throwable) -> semaphore.release());
+ } catch (InterruptedException e) {
+ semaphore.release();
+ }
+ }
+
+ @Override
+ public void dispatch(DispatchRequest request) {
+ if (stopped || topicFilter != null && topicFilter.filterTopic(request.getTopic())) {
+ return;
+ }
+ flatFileStore.computeIfAbsent(
+ new MessageQueue(request.getTopic(), brokerName, request.getQueueId()));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> dispatchAsync(FlatFileInterface flatFile, boolean force) {
+ if (stopped) {
+ return CompletableFuture.completedFuture(true);
+ }
+
+ String topic = flatFile.getMessageQueue().getTopic();
+ int queueId = flatFile.getMessageQueue().getQueueId();
+
+ force = !storeConfig.isTieredStoreGroupCommit() || force;
+ if (force) {
+ flatFile.getFileLock().lock();
+ } else {
+ if (!flatFile.getFileLock().tryLock()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ }
+
+ try {
+ if (topicFilter != null && topicFilter.filterTopic(flatFile.getMessageQueue().getTopic())) {
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long currentOffset = flatFile.getConsumeQueueMaxOffset();
+ long commitOffset = flatFile.getConsumeQueueCommitOffset();
+ long minOffsetInQueue = defaultStore.getMinOffsetInQueue(topic, queueId);
+ long maxOffsetInQueue = defaultStore.getMaxOffsetInQueue(topic, queueId);
+
+ // If set to max offset here, some written messages may be lost
+ if (!flatFile.isFlatFileInit()) {
+ currentOffset = Math.max(minOffsetInQueue,
+ maxOffsetInQueue - storeConfig.getTieredStoreGroupCommitSize());
+ flatFile.initOffset(currentOffset);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ // If last commit failed
+ if (commitOffset < currentOffset) {
+ this.commitAsync(flatFile);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset == maxOffsetInQueue) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset < minOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too small, " +
+ "topic={}, queueId={}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ flatFileStore.computeIfAbsent(new MessageQueue(topic, brokerName, queueId));
+ return CompletableFuture.completedFuture(true);
+ }
+
+ if (currentOffset > maxOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too large, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long interval = TimeUnit.HOURS.toMillis(storeConfig.getCommitLogRollingInterval());
+ if (flatFile.rollingFile(interval)) {
+ log.warn("MessageDispatcher#dispatch, rolling file, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ }
+
+ long bufferSize = 0L;
+ long groupCommitSize = storeConfig.getTieredStoreGroupCommitSize();
+ long groupCommitCount = storeConfig.getTieredStoreGroupCommitCount();
+ long targetOffset = Math.min(currentOffset + groupCommitCount, maxOffsetInQueue);
+
+ ConsumeQueueInterface consumeQueue = defaultStore.getConsumeQueue(topic, queueId);
+ CqUnit cqUnit = consumeQueue.get(currentOffset);
+ SelectMappedBufferResult message =
+ defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
+ boolean timeout = MessageFormatUtil.getStoreTimeStamp(message.getByteBuffer()) +
+ TimeUnit.SECONDS.toMillis(30) < System.currentTimeMillis();
Review Comment:
谢谢，另外想再多问一句dfs在这里指的是什么
--
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 #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "lizhimins (via GitHub)" <gi...@apache.org>.
lizhimins merged PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899
--
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 #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "bxfjb (via GitHub)" <gi...@apache.org>.
bxfjb commented on code in PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899#discussion_r1525862549
##########
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.tieredstore.core;
+
+import io.opentelemetry.api.common.Attributes;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
+import org.apache.rocketmq.store.queue.CqUnit;
+import org.apache.rocketmq.tieredstore.MessageStoreConfig;
+import org.apache.rocketmq.tieredstore.MessageStoreExecutor;
+import org.apache.rocketmq.tieredstore.TieredMessageStore;
+import org.apache.rocketmq.tieredstore.common.AppendResult;
+import org.apache.rocketmq.tieredstore.common.FileSegmentType;
+import org.apache.rocketmq.tieredstore.file.FlatFileInterface;
+import org.apache.rocketmq.tieredstore.file.FlatFileStore;
+import org.apache.rocketmq.tieredstore.index.IndexService;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
+import org.apache.rocketmq.tieredstore.util.MessageFormatUtil;
+import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MessageStoreDispatcherImpl extends ServiceThread implements MessageStoreDispatcher {
+
+ protected static final Logger log = LoggerFactory.getLogger(MessageStoreUtil.TIERED_STORE_LOGGER_NAME);
+
+ protected final String brokerName;
+ protected final MessageStore defaultStore;
+ protected final MessageStoreConfig storeConfig;
+ protected final TieredMessageStore messageStore;
+ protected final FlatFileStore flatFileStore;
+ protected final MessageStoreExecutor storeExecutor;
+ protected final MessageStoreFilter topicFilter;
+ protected final Semaphore semaphore;
+ protected final IndexService indexService;
+
+ public MessageStoreDispatcherImpl(TieredMessageStore messageStore) {
+ this.messageStore = messageStore;
+ this.storeConfig = messageStore.getStoreConfig();
+ this.defaultStore = messageStore.getDefaultStore();
+ this.brokerName = storeConfig.getBrokerName();
+ this.semaphore = new Semaphore(
+ this.storeConfig.getTieredStoreMaxPendingLimit() / 4);
+ this.topicFilter = messageStore.getTopicFilter();
+ this.flatFileStore = messageStore.getFlatFileStore();
+ this.storeExecutor = messageStore.getStoreExecutor();
+ this.indexService = messageStore.getIndexService();
+ }
+
+ @Override
+ public String getServiceName() {
+ return MessageStoreDispatcher.class.getSimpleName();
+ }
+
+ public void dispatchWithSemaphore(FlatFileInterface flatFile) {
+ try {
+ if (stopped) {
+ return;
+ }
+ semaphore.acquire();
+ this.dispatchAsync(flatFile, false)
+ .whenComplete((future, throwable) -> semaphore.release());
+ } catch (InterruptedException e) {
+ semaphore.release();
+ }
+ }
+
+ @Override
+ public void dispatch(DispatchRequest request) {
+ if (stopped || topicFilter != null && topicFilter.filterTopic(request.getTopic())) {
+ return;
+ }
+ flatFileStore.computeIfAbsent(
+ new MessageQueue(request.getTopic(), brokerName, request.getQueueId()));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> dispatchAsync(FlatFileInterface flatFile, boolean force) {
+ if (stopped) {
+ return CompletableFuture.completedFuture(true);
+ }
+
+ String topic = flatFile.getMessageQueue().getTopic();
+ int queueId = flatFile.getMessageQueue().getQueueId();
+
+ force = !storeConfig.isTieredStoreGroupCommit() || force;
+ if (force) {
+ flatFile.getFileLock().lock();
+ } else {
+ if (!flatFile.getFileLock().tryLock()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ }
+
+ try {
+ if (topicFilter != null && topicFilter.filterTopic(flatFile.getMessageQueue().getTopic())) {
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long currentOffset = flatFile.getConsumeQueueMaxOffset();
+ long commitOffset = flatFile.getConsumeQueueCommitOffset();
+ long minOffsetInQueue = defaultStore.getMinOffsetInQueue(topic, queueId);
+ long maxOffsetInQueue = defaultStore.getMaxOffsetInQueue(topic, queueId);
+
+ // If set to max offset here, some written messages may be lost
+ if (!flatFile.isFlatFileInit()) {
+ currentOffset = Math.max(minOffsetInQueue,
+ maxOffsetInQueue - storeConfig.getTieredStoreGroupCommitSize());
+ flatFile.initOffset(currentOffset);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ // If last commit failed
+ if (commitOffset < currentOffset) {
+ this.commitAsync(flatFile);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset == maxOffsetInQueue) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset < minOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too small, " +
+ "topic={}, queueId={}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ flatFileStore.computeIfAbsent(new MessageQueue(topic, brokerName, queueId));
+ return CompletableFuture.completedFuture(true);
+ }
+
+ if (currentOffset > maxOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too large, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long interval = TimeUnit.HOURS.toMillis(storeConfig.getCommitLogRollingInterval());
+ if (flatFile.rollingFile(interval)) {
+ log.warn("MessageDispatcher#dispatch, rolling file, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ }
+
+ long bufferSize = 0L;
+ long groupCommitSize = storeConfig.getTieredStoreGroupCommitSize();
+ long groupCommitCount = storeConfig.getTieredStoreGroupCommitCount();
+ long targetOffset = Math.min(currentOffset + groupCommitCount, maxOffsetInQueue);
+
+ ConsumeQueueInterface consumeQueue = defaultStore.getConsumeQueue(topic, queueId);
+ CqUnit cqUnit = consumeQueue.get(currentOffset);
+ SelectMappedBufferResult message =
+ defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
+ boolean timeout = MessageFormatUtil.getStoreTimeStamp(message.getByteBuffer()) +
+ TimeUnit.SECONDS.toMillis(30) < System.currentTimeMillis();
Review Comment:
谢谢，另外想再多问一句dfs在这里指的是什么
--
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 #7878] Performance Improvement and Bug Fixes for the Tiered Storage Module [rocketmq]
Posted by "bxfjb (via GitHub)" <gi...@apache.org>.
bxfjb commented on code in PR #7899:
URL: https://github.com/apache/rocketmq/pull/7899#discussion_r1525862549
##########
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImpl.java:
##########
@@ -0,0 +1,297 @@
+/*
+ * 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.tieredstore.core;
+
+import io.opentelemetry.api.common.Attributes;
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.common.ServiceThread;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.store.DispatchRequest;
+import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.SelectMappedBufferResult;
+import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
+import org.apache.rocketmq.store.queue.CqUnit;
+import org.apache.rocketmq.tieredstore.MessageStoreConfig;
+import org.apache.rocketmq.tieredstore.MessageStoreExecutor;
+import org.apache.rocketmq.tieredstore.TieredMessageStore;
+import org.apache.rocketmq.tieredstore.common.AppendResult;
+import org.apache.rocketmq.tieredstore.common.FileSegmentType;
+import org.apache.rocketmq.tieredstore.file.FlatFileInterface;
+import org.apache.rocketmq.tieredstore.file.FlatFileStore;
+import org.apache.rocketmq.tieredstore.index.IndexService;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsConstant;
+import org.apache.rocketmq.tieredstore.metrics.TieredStoreMetricsManager;
+import org.apache.rocketmq.tieredstore.util.MessageFormatUtil;
+import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MessageStoreDispatcherImpl extends ServiceThread implements MessageStoreDispatcher {
+
+ protected static final Logger log = LoggerFactory.getLogger(MessageStoreUtil.TIERED_STORE_LOGGER_NAME);
+
+ protected final String brokerName;
+ protected final MessageStore defaultStore;
+ protected final MessageStoreConfig storeConfig;
+ protected final TieredMessageStore messageStore;
+ protected final FlatFileStore flatFileStore;
+ protected final MessageStoreExecutor storeExecutor;
+ protected final MessageStoreFilter topicFilter;
+ protected final Semaphore semaphore;
+ protected final IndexService indexService;
+
+ public MessageStoreDispatcherImpl(TieredMessageStore messageStore) {
+ this.messageStore = messageStore;
+ this.storeConfig = messageStore.getStoreConfig();
+ this.defaultStore = messageStore.getDefaultStore();
+ this.brokerName = storeConfig.getBrokerName();
+ this.semaphore = new Semaphore(
+ this.storeConfig.getTieredStoreMaxPendingLimit() / 4);
+ this.topicFilter = messageStore.getTopicFilter();
+ this.flatFileStore = messageStore.getFlatFileStore();
+ this.storeExecutor = messageStore.getStoreExecutor();
+ this.indexService = messageStore.getIndexService();
+ }
+
+ @Override
+ public String getServiceName() {
+ return MessageStoreDispatcher.class.getSimpleName();
+ }
+
+ public void dispatchWithSemaphore(FlatFileInterface flatFile) {
+ try {
+ if (stopped) {
+ return;
+ }
+ semaphore.acquire();
+ this.dispatchAsync(flatFile, false)
+ .whenComplete((future, throwable) -> semaphore.release());
+ } catch (InterruptedException e) {
+ semaphore.release();
+ }
+ }
+
+ @Override
+ public void dispatch(DispatchRequest request) {
+ if (stopped || topicFilter != null && topicFilter.filterTopic(request.getTopic())) {
+ return;
+ }
+ flatFileStore.computeIfAbsent(
+ new MessageQueue(request.getTopic(), brokerName, request.getQueueId()));
+ }
+
+ @Override
+ public CompletableFuture<Boolean> dispatchAsync(FlatFileInterface flatFile, boolean force) {
+ if (stopped) {
+ return CompletableFuture.completedFuture(true);
+ }
+
+ String topic = flatFile.getMessageQueue().getTopic();
+ int queueId = flatFile.getMessageQueue().getQueueId();
+
+ force = !storeConfig.isTieredStoreGroupCommit() || force;
+ if (force) {
+ flatFile.getFileLock().lock();
+ } else {
+ if (!flatFile.getFileLock().tryLock()) {
+ return CompletableFuture.completedFuture(false);
+ }
+ }
+
+ try {
+ if (topicFilter != null && topicFilter.filterTopic(flatFile.getMessageQueue().getTopic())) {
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long currentOffset = flatFile.getConsumeQueueMaxOffset();
+ long commitOffset = flatFile.getConsumeQueueCommitOffset();
+ long minOffsetInQueue = defaultStore.getMinOffsetInQueue(topic, queueId);
+ long maxOffsetInQueue = defaultStore.getMaxOffsetInQueue(topic, queueId);
+
+ // If set to max offset here, some written messages may be lost
+ if (!flatFile.isFlatFileInit()) {
+ currentOffset = Math.max(minOffsetInQueue,
+ maxOffsetInQueue - storeConfig.getTieredStoreGroupCommitSize());
+ flatFile.initOffset(currentOffset);
+ return CompletableFuture.completedFuture(true);
+ }
+
+ // If last commit failed
+ if (commitOffset < currentOffset) {
+ this.commitAsync(flatFile);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset == maxOffsetInQueue) {
+ return CompletableFuture.completedFuture(false);
+ }
+
+ if (currentOffset < minOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too small, " +
+ "topic={}, queueId={}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ flatFileStore.destroyFile(flatFile.getMessageQueue());
+ flatFileStore.computeIfAbsent(new MessageQueue(topic, brokerName, queueId));
+ return CompletableFuture.completedFuture(true);
+ }
+
+ if (currentOffset > maxOffsetInQueue) {
+ log.warn("MessageDispatcher#dispatch, current offset is too large, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ return CompletableFuture.completedFuture(false);
+ }
+
+ long interval = TimeUnit.HOURS.toMillis(storeConfig.getCommitLogRollingInterval());
+ if (flatFile.rollingFile(interval)) {
+ log.warn("MessageDispatcher#dispatch, rolling file, " +
+ "topic: {}, queueId: {}, offset={}-{}, current={}",
+ topic, queueId, minOffsetInQueue, maxOffsetInQueue, currentOffset);
+ }
+
+ long bufferSize = 0L;
+ long groupCommitSize = storeConfig.getTieredStoreGroupCommitSize();
+ long groupCommitCount = storeConfig.getTieredStoreGroupCommitCount();
+ long targetOffset = Math.min(currentOffset + groupCommitCount, maxOffsetInQueue);
+
+ ConsumeQueueInterface consumeQueue = defaultStore.getConsumeQueue(topic, queueId);
+ CqUnit cqUnit = consumeQueue.get(currentOffset);
+ SelectMappedBufferResult message =
+ defaultStore.selectOneMessageByOffset(cqUnit.getPos(), cqUnit.getSize());
+ boolean timeout = MessageFormatUtil.getStoreTimeStamp(message.getByteBuffer()) +
+ TimeUnit.SECONDS.toMillis(30) < System.currentTimeMillis();
Review Comment:
谢谢另外想再多问一句dfs在这里指的是什么
--
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