You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "lollipopjin (via GitHub)" <gi...@apache.org> on 2023/12/04 07:10:12 UTC
Re: [PR] [ISSUE #7585] Support message filtering in rocketmq tiered storage [rocketmq]
lollipopjin commented on code in PR #7594:
URL: https://github.com/apache/rocketmq/pull/7594#discussion_r1413322104
##########
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java:
##########
@@ -303,74 +270,94 @@ public CompletableFuture<GetMessageResult> getMessageFromCacheAsync(CompositeQue
recordCacheAccess(flatFile, group, queueOffset, resultWrapperList);
- // if cache hit, result will be returned immediately and asynchronously prefetch messages for later requests
- if (!resultWrapperList.isEmpty()) {
- LOGGER.debug("MessageFetcher#getMessageFromCacheAsync: cache hit: " +
- "topic: {}, queue: {}, queue offset: {}, max message num: {}, cache hit num: {}",
- mq.getTopic(), mq.getQueueId(), queueOffset, maxCount, resultWrapperList.size());
- prefetchMessage(flatFile, group, maxCount, lastGetOffset + 1);
+ if (resultWrapperList.isEmpty()) {
+ // If cache miss, pull messages immediately
+ LOGGER.info("MessageFetcher cache miss, group: {}, topic: {}, queueId: {}, offset: {}, maxCount: {}",
+ group, mq.getTopic(), mq.getQueueId(), queueOffset, maxCount);
+ } else {
+ // If cache hit, return buffer result immediately and asynchronously prefetch messages
+ LOGGER.debug("MessageFetcher cache hit, group: {}, topic: {}, queueId: {}, offset: {}, maxCount: {}, resultSize: {}",
+ group, mq.getTopic(), mq.getQueueId(), queueOffset, maxCount, resultWrapperList.size());
- GetMessageResult result = new GetMessageResult();
+ GetMessageResultExt result = new GetMessageResultExt();
result.setStatus(GetMessageStatus.FOUND);
result.setMinOffset(flatFile.getConsumeQueueMinOffset());
result.setMaxOffset(flatFile.getConsumeQueueCommitOffset());
result.setNextBeginOffset(queueOffset + resultWrapperList.size());
- resultWrapperList.forEach(wrapper -> result.addMessage(wrapper.getDuplicateResult(), wrapper.getCurOffset()));
+ resultWrapperList.forEach(wrapper -> result.addMessageExt(
+ wrapper.getDuplicateResult(), wrapper.getOffset(), wrapper.getTagCode()));
+
+ if (lastGetOffset < result.getMaxOffset()) {
+ this.prefetchMessage(flatFile, group, maxCount, lastGetOffset + 1);
+ }
return CompletableFuture.completedFuture(result);
}
- // if cache is miss, immediately pull messages
- LOGGER.info("TieredMessageFetcher#getMessageFromCacheAsync: cache miss: " +
- "topic: {}, queue: {}, queue offset: {}, max message num: {}",
- mq.getTopic(), mq.getQueueId(), queueOffset, maxCount);
-
- CompletableFuture<GetMessageResult> resultFuture;
+ CompletableFuture<GetMessageResultExt> resultFuture;
synchronized (flatFile) {
int batchSize = maxCount * storeConfig.getReadAheadMinFactor();
resultFuture = getMessageFromTieredStoreAsync(flatFile, queueOffset, batchSize)
- .thenApplyAsync(result -> {
+ .thenApply(result -> {
if (result.getStatus() != GetMessageStatus.FOUND) {
return result;
}
- GetMessageResult newResult = new GetMessageResult();
- newResult.setStatus(GetMessageStatus.FOUND);
- newResult.setMinOffset(flatFile.getConsumeQueueMinOffset());
- newResult.setMaxOffset(flatFile.getConsumeQueueCommitOffset());
+ GetMessageResultExt newResult = new GetMessageResultExt();
List<Long> offsetList = result.getMessageQueueOffset();
+ List<Long> tagCodeList = result.getTagCodeList();
List<SelectMappedBufferResult> msgList = result.getMessageMapedList();
- Long minOffset = offsetList.get(0);
- Long maxOffset = offsetList.get(offsetList.size() - 1);
- int size = offsetList.size();
+
for (int i = 0; i < offsetList.size(); i++) {
- Long offset = offsetList.get(i);
SelectMappedBufferResult msg = msgList.get(i);
- // put message into cache
- SelectMappedBufferResultWrapper resultWrapper = putMessageToCache(flatFile, offset, msg, minOffset, maxOffset, size, true);
- // try to meet maxCount
+ SelectBufferResultWrapper bufferResult = new SelectBufferResultWrapper(
+ msg, offsetList.get(i), tagCodeList.get(i), true);
+ this.putMessageToCache(flatFile, bufferResult);
if (newResult.getMessageMapedList().size() < maxCount) {
- newResult.addMessage(resultWrapper.getDuplicateResult(), offset);
+ newResult.addMessageExt(msg, offsetList.get(i), tagCodeList.get(i));
}
}
+
+ newResult.setStatus(GetMessageStatus.FOUND);
+ newResult.setMinOffset(flatFile.getConsumeQueueMinOffset());
+ newResult.setMaxOffset(flatFile.getConsumeQueueCommitOffset());
newResult.setNextBeginOffset(queueOffset + newResult.getMessageMapedList().size());
return newResult;
- }, TieredStoreExecutor.fetchDataExecutor);
+ });
List<Pair<Integer, CompletableFuture<Long>>> futureList = new ArrayList<>();
CompletableFuture<Long> inflightRequestFuture = resultFuture.thenApply(result ->
- result.getStatus() == GetMessageStatus.FOUND ? result.getMessageQueueOffset().get(result.getMessageQueueOffset().size() - 1) : -1L);
+ result.getStatus() == GetMessageStatus.FOUND ?
+ result.getMessageQueueOffset().get(result.getMessageQueueOffset().size() - 1) : -1L);
Review Comment:
How about extract "result.getMessageQueueOffset().size() - 1" as lastQueueOffsetElementIndex ?
--
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