Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
HScarb merged 5 commits into from
Sep 2, 2025
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
914491a
[ISSUE #9648] Fix getOffsetInQueueByTime missing in tieredMessageStore
GabrielDUX Aug 30, 2025
c04ea31
Update TieredMessageStoreTest.java
GabrielDUX Aug 30, 2025
5d7dae9
Merge branch 'apache:develop' into fix_boundryType_missing_tieredStore
GabrielDUX Sep 1, 2025
79ae6d0
Delete inappropriate UT
GabrielDUX Sep 2, 2025
e325aa4
Remove unused import
GabrielDUX Sep 2, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 0 additions & 38 deletions store/src/test/java/org/apache/rocketmq/store/MessageStoreStateMachineTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

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;
Expand Down Expand Up @@ -107,41 +106,4 @@ public void testGetCurrentState() {
// 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);
}
}
4 changes: 2 additions & 2 deletions tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -370,11 +370,11 @@ public long getOffsetInQueueByTime(String topic, int queueId, long timestamp, Bo
.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
Expand Down
33 changes: 28 additions & 5 deletions tieredstore/src/test/java/org/apache/rocketmq/tieredstore/TieredMessageStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,6 @@
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 {
Expand Down Expand Up @@ -275,19 +274,43 @@ public void testGetMessageStoreTimeStampAsync() {

@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);
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);

// 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));

Mockito.when(fetcher.getOffsetInQueueByTime(anyString(), anyInt(), anyLong(), eq(BoundaryType.LOWER))).thenReturn(-1L);
// 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
Expand Down
Loading