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

[ISSUE #9304] Resolve cold data read control issue in DefaultMessageStore (#9305)

Browse Source
This commit is contained in:
ymwneu
2025-07-19 18:17:56 +08:00
committed by RongtongJin
parent dd1d5b99e0
commit f818df501c
1 changed files with 7 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+7 -5
View File
@@ -64,6 +64,9 @@ import org.apache.rocketmq.store.ha.HAService;
import org.apache.rocketmq.store.ha.autoswitch.AutoSwitchHAService;
import org.apache.rocketmq.store.lock.AdaptiveBackOffSpinLockImpl;
import org.apache.rocketmq.store.logfile.MappedFile;
import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
import org.apache.rocketmq.store.queue.CqUnit;
import org.apache.rocketmq.store.queue.ReferredIterator;
import org.apache.rocketmq.store.util.LibC;
import org.rocksdb.RocksDBException;
@@ -2442,16 +2445,15 @@ public class CommitLog implements Swappable {
return false;
}
try {
ConsumeQueue consumeQueue = (ConsumeQueue) defaultMessageStore.findConsumeQueue(topic, queueId);
ConsumeQueueInterface consumeQueue = defaultMessageStore.findConsumeQueue(topic, queueId);
if (null == consumeQueue) {
return false;
}
SelectMappedBufferResult bufferConsumeQueue = consumeQueue.getIndexBuffer(offset);
if (null == bufferConsumeQueue || null == bufferConsumeQueue.getByteBuffer()) {
ReferredIterator<CqUnit> bufferConsumeQueue = consumeQueue.iterateFrom(offset, 1);
if (null == bufferConsumeQueue || !bufferConsumeQueue.hasNext()) {
return false;
}
long offsetPy = bufferConsumeQueue.getByteBuffer().getLong();
return defaultMessageStore.checkInColdAreaByCommitOffset(offsetPy, getMaxOffset());
return defaultMessageStore.checkInColdAreaByCommitOffset(bufferConsumeQueue.next().getPos(), getMaxOffset());
} catch (Exception e) {
log.error("isMsgInColdArea group: {}, topic: {}, queueId: {}, offset: {}",
group, topic, queueId, offset, e);