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

[ISSUE #10268] Fix incorrect time range file selection in IndexStoreService.queryAsync (#10269)

Browse Source 
Co-authored-by: lizhimins <lizhimins@users.noreply.github.com>
This commit is contained in:
lizhimins
2026-04-21 19:15:26 +08:00
committed by GitHub
parent 840b18c454
commit 94fbfcf342
2 changed files with 41 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreService.java
+4 -1
View File
@@ -235,11 +235,14 @@ public class IndexStoreService extends ServiceThread implements IndexService {
try {
readWriteLock.readLock().lock();
ConcurrentNavigableMap<Long, IndexFile> pendingMap =
this.timeStoreTable.subMap(beginTime, true, endTime, true);
this.timeStoreTable.headMap(endTime, true);
List<CompletableFuture<Void>> futureList = new ArrayList<>(pendingMap.size());
ConcurrentSkipListMap<String /* queueId-offset */, IndexItem> result = new ConcurrentSkipListMap<>();
for (Map.Entry<Long, IndexFile> entry : pendingMap.descendingMap().entrySet()) {
if (entry.getValue().getEndTimestamp() < beginTime) {
break;
}
CompletableFuture<Void> completableFuture = entry.getValue()
.queryAsync(topic, key, maxCount, beginTime, endTime)
.thenAccept(itemList -> itemList.forEach(indexItem -> {
tieredstore/src/test/java/org/apache/rocketmq/tieredstore/index/IndexStoreServiceTest.java
+37
View File
@@ -351,4 +351,41 @@ public class IndexStoreServiceTest {
executorService.shutdown();
Assert.assertTrue(result.get());
}
@Test
public void queryCrossFileBoundaryTest() throws InterruptedException, ExecutionException {
indexService = new IndexStoreService(fileAllocator, filePath);
indexService.start();
// Create first file with early beginTime
long file1Begin = System.currentTimeMillis();
for (int i = 0; i < storeConfig.getTieredStoreIndexFileMaxIndexNum() - 1; i++) {
indexService.putKey(TOPIC_NAME, TOPIC_ID, QUEUE_ID,
Collections.singleton("crossKey"), i * 100L, MESSAGE_SIZE, file1Begin + i * 1000);
}
// Create second file with later beginTime (beyond query range)
long file2Begin = System.currentTimeMillis() + 100_000;
indexService.createNewIndexFile(file2Begin);
for (int i = 0; i < 5; i++) {
indexService.putKey(TOPIC_NAME, TOPIC_ID, QUEUE_ID,
Collections.singleton("crossKey"), i * 100L, MESSAGE_SIZE, file2Begin + i);
}
Assert.assertEquals(2, indexService.getTimeStoreTable().size());
// Query range: beginTime is AFTER file1's beginTime but BEFORE file1's last item timestamp
// This should select file1, NOT file2 (file2 beginTime > queryEnd)
long queryBegin = file1Begin + 5_000;
long queryEnd = file1Begin + 15_000;
List<IndexItem> results = indexService.queryAsync(
TOPIC_NAME, "crossKey", 10, queryBegin, queryEnd).get();
// file1 has items at timestamps: file1Begin, file1Begin+1000, ..., file1Begin+(N-1)*1000
// Items in range [file1Begin+5000, file1Begin+15000] should match
// The bug (subMap) would return empty because file1's key < queryBegin
Assert.assertFalse("Should find index items from file covering query range", results.isEmpty());
Assert.assertTrue("Should find items within query time range", results.size() > 0);
}
}