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

[ISSUE #9648] Fix getOffsetInQueueByTime missing boundaryType in tieredMessageStore (#9649)

Browse Source 
* [ISSUE #9648] Fix getOffsetInQueueByTime missing in tieredMessageStore

* Update TieredMessageStoreTest.java

* Delete inappropriate UT

* Remove unused import
This commit is contained in:
Duxuwei
2025-09-02 19:49:40 +08:00
committed by GitHub
parent 29546558d2
commit 74ab3ae112
3 changed files with 32 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/test/java/org/apache/rocketmq/store/MessageStoreStateMachineTest.java
-38
View File
@@ -19,7 +19,6 @@ package org.apache.rocketmq.store;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.anyLong;
@@ -107,41 +106,4 @@ public class MessageStoreStateMachineTest {
// Verify the current state
assertEquals(MessageStoreState.INIT, stateMachine.getCurrentState());
}
/**
* Test getTotalRunningTimeMs method.
*/
@Test
public void testGetTotalRunningTimeMs() {
// Sleep for a short duration to simulate elapsed time
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// Verify the total running time is approximately correct
long totalTime = stateMachine.getTotalRunningTimeMs();
assertTrue(totalTime >= 100 && totalTime < 200);
}
/**
* Test getCurrentStateRunningTimeMs method.
*/
@Test
public void testGetCurrentStateRunningTimeMs() {
// Perform a state transition
stateMachine.transitTo(MessageStoreState.LOAD_COMMITLOG_OK);
// Sleep for a short duration to simulate elapsed time
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// Verify the current state running time is approximately correct
long currentStateTime = stateMachine.getCurrentStateRunningTimeMs();
assertTrue(currentStateTime >= 100 && currentStateTime < 200);
}
}
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java
+2 -2
View File
@@ -370,11 +370,11 @@ public class TieredMessageStore extends AbstractPluginMessageStore {
.build();
TieredStoreMetricsManager.apiLatency.record(stopwatch.elapsed(TimeUnit.MILLISECONDS), latencyAttributes);
if (offsetInTieredStore == -1L && !isForce) {
return next.getOffsetInQueueByTime(topic, queueId, timestamp);
return next.getOffsetInQueueByTime(topic, queueId, timestamp, boundaryType);
}
return offsetInTieredStore;
}
return next.getOffsetInQueueByTime(topic, queueId, timestamp);
return next.getOffsetInQueueByTime(topic, queueId, timestamp, boundaryType);
}
@Override
tieredstore/src/test/java/org/apache/rocketmq/tieredstore/TieredMessageStoreTest.java
+30 -7
View File
@@ -62,7 +62,6 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.when;
public class TieredMessageStoreTest {
@@ -275,19 +274,43 @@ public class TieredMessageStoreTest {
@Test
public void testGetOffsetInQueueByTime() {
final long earliestMsgTime = 100L;
Properties properties = new Properties();
properties.setProperty("tieredStorageLevel", "FORCE");
configuration.update(properties);
Mockito.when(fetcher.getOffsetInQueueByTime(anyString(), anyInt(), anyLong(), eq(BoundaryType.LOWER))).thenReturn(1L);
Mockito.when(defaultStore.getOffsetInQueueByTime(anyString(), anyInt(), anyLong())).thenReturn(2L);
Mockito.when(defaultStore.getEarliestMessageTime()).thenReturn(100L);
Assert.assertEquals(1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 1000, BoundaryType.LOWER));
Assert.assertEquals(1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 0, BoundaryType.LOWER));
Mockito.when(fetcher.getOffsetInQueueByTime(anyString(), anyInt(), anyLong(), any(BoundaryType.class)))
.thenAnswer(ivk -> ivk.getArgument(3, BoundaryType.class) == BoundaryType.LOWER ? 1L : 2L);
Mockito.when(defaultStore.getOffsetInQueueByTime(anyString(), anyInt(), anyLong(), any(BoundaryType.class)))
.thenAnswer(ivk -> {
long time = ivk.getArgument(2, Long.class);
if (time < earliestMsgTime) {
return -1L;
}
return ivk.getArgument(3, BoundaryType.class) == BoundaryType.LOWER ? 3L : 4L;
});
Mockito.when(defaultStore.getEarliestMessageTime()).thenReturn(earliestMsgTime);
Mockito.when(fetcher.getOffsetInQueueByTime(anyString(), anyInt(), anyLong(), eq(BoundaryType.LOWER))).thenReturn(-1L);
// Message not in disk, but force, found in tired storage.
Assert.assertEquals(1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 1000, BoundaryType.LOWER));
Assert.assertEquals(2L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 1000, BoundaryType.UPPER));
// Message in disk, and force, found in tired storage.
Assert.assertEquals(1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 0, BoundaryType.LOWER));
Assert.assertEquals(2L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 0, BoundaryType.UPPER));
// Message in disk, but force, and not found in tired storage.
Mockito.when(fetcher.getOffsetInQueueByTime(anyString(), anyInt(), anyLong(), any(BoundaryType.class))).thenReturn(-1L);
Assert.assertEquals(-1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 0));
Assert.assertEquals(-1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 0, BoundaryType.LOWER));
properties.setProperty("tieredStorageLevel", "NOT_IN_DISK");
configuration.update(properties);
// Message not in disk, and not found in tired storage.
Assert.assertEquals(-1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 0, BoundaryType.LOWER));
Assert.assertEquals(-1L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 0, BoundaryType.UPPER));
// Message in disk, and found in disk.
Assert.assertEquals(3L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 1000, BoundaryType.LOWER));
Assert.assertEquals(4L, currentStore.getOffsetInQueueByTime(mq.getTopic(), mq.getQueueId(), 1000, BoundaryType.UPPER));
}
@Test