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

[ISSUE #10201] Optimize queryOffset method overloads in IndexService (#10202)

Browse Source
This commit is contained in:
yx9o
2026-03-25 15:24:21 +08:00
committed by GitHub
parent 59033a3653
commit 932588d3e1
2 changed files with 50 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/index/IndexService.java
+3 -38
View File
@@ -167,49 +167,14 @@ public class IndexService implements CommitLogDispatchStore {
}
public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end) {
long indexLastUpdateTimestamp = 0;
long indexLastUpdatePhyoffset = 0;
maxNum = Math.min(maxNum, this.defaultMessageStore.getMessageStoreConfig().getMaxMsgsNumBatch());
List<Long> phyOffsets = new ArrayList<>(maxNum);
try {
this.readWriteLock.readLock().lock();
if (!this.indexFileList.isEmpty()) {
for (int i = this.indexFileList.size(); i > 0; i--) {
IndexFile f = this.indexFileList.get(i - 1);
boolean lastFile = i == this.indexFileList.size();
if (lastFile) {
indexLastUpdateTimestamp = f.getEndTimestamp();
indexLastUpdatePhyoffset = f.getEndPhyOffset();
}
if (f.isTimeMatched(begin, end)) {
f.selectPhyOffset(phyOffsets, buildKey(topic, key), maxNum, begin, end);
}
if (f.getBeginTimestamp() < begin) {
break;
}
if (phyOffsets.size() >= maxNum) {
break;
}
}
}
} catch (Exception e) {
LOGGER.error("queryMsg exception", e);
} finally {
this.readWriteLock.readLock().unlock();
}
return new QueryOffsetResult(phyOffsets, indexLastUpdateTimestamp, indexLastUpdatePhyoffset);
return queryOffset(topic, key, maxNum, begin, end, null);
}
public QueryOffsetResult queryOffset(String topic, String key, int maxNum, long begin, long end, String indexType) {
List<Long> phyOffsets = new ArrayList<>(maxNum);
long indexLastUpdateTimestamp = 0;
long indexLastUpdatePhyoffset = 0;
maxNum = Math.min(maxNum, this.defaultMessageStore.getMessageStoreConfig().getMaxMsgsNumBatch());
List<Long> phyOffsets = new ArrayList<>(maxNum);
try {
this.readWriteLock.readLock().lock();
if (!this.indexFileList.isEmpty()) {
@@ -241,7 +206,7 @@ public class IndexService implements CommitLogDispatchStore {
}
}
} catch (Exception e) {
LOGGER.error("queryMsg queryOffset exception", e);
LOGGER.error("queryOffset exception", e);
} finally {
this.readWriteLock.readLock().unlock();
}
store/src/test/java/org/apache/rocketmq/store/index/IndexServiceTest.java
+47 -12
View File
@@ -21,29 +21,64 @@ import org.apache.rocketmq.common.BrokerConfig;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.apache.rocketmq.store.stats.BrokerStatsManager;
import org.junit.Before;
import org.junit.Test;
import java.util.Collections;
import java.util.concurrent.ConcurrentHashMap;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
public class IndexServiceTest {
@Test
public void testQueryOffsetThrow() throws Exception {
assertDoesNotThrow(() -> {
DefaultMessageStore store = new DefaultMessageStore(
new MessageStoreConfig(),
new BrokerStatsManager(new BrokerConfig()),
null,
new BrokerConfig(),
new ConcurrentHashMap<>()
);
private IndexService indexService;
IndexService indexService = new IndexService(store);
@Before
public void setUp() throws Exception {
DefaultMessageStore store = new DefaultMessageStore(
new MessageStoreConfig(),
new BrokerStatsManager(new BrokerConfig()),
null,
new BrokerConfig(),
new ConcurrentHashMap<>()
);
indexService = new IndexService(store);
}
@Test
public void testQueryOffsetThrow() {
assertDoesNotThrow(() -> {
indexService.queryOffset("test", "", Integer.MAX_VALUE, 10, 100);
});
}
@Test
public void testQueryOffsetWithoutIndexType() {
QueryOffsetResult result = indexService.queryOffset("test", "testKey", 10, 0, 100);
assertNotNull(result);
assertEquals(Collections.emptyList(), result.getPhyOffsets());
}
@Test
public void testQueryOffsetWithIndexType() {
QueryOffsetResult result = indexService.queryOffset("test", "testKey", 10, 0, 100, "TAG");
assertNotNull(result);
assertEquals(Collections.emptyList(), result.getPhyOffsets());
}
@Test
public void testQueryOffsetWithNullKey() {
QueryOffsetResult result = indexService.queryOffset("test", null, 10, 0, 100);
assertNotNull(result);
assertEquals(Collections.emptyList(), result.getPhyOffsets());
}
@Test
public void testQueryOffsetWithZeroMaxNum() {
QueryOffsetResult result = indexService.queryOffset("test", "testKey", 0, 0, 100);
assertNotNull(result);
assertEquals(Collections.emptyList(), result.getPhyOffsets());
}
}