This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new 4babef3e6 fix(message): validate the time window of a message-key and
unique-key lookup (#4717)
4babef3e6 is described below
commit 4babef3e67f28cde16648568d949666598f37208
Author: btlqql <[email protected]>
AuthorDate: Sat Sep 26 18:48:21 2026 +0800
fix(message): validate the time window of a message-key and unique-key
lookup (#4717)
fix(message): validate the time window of a message-key query
A key/unique-key lookup may carry an optional window, but MessageService
forwarded it
to the provider unvalidated: validateTopicQueryWindow returns early for a
key and
queryMessageByUniqueKey had no check at all, so a negative or inverted
window reached
the broker / cloud API and the response depended on the provider. The
window is now
the same 400 everywhere, without the seven-day bound that belongs to topic
scans.
---
.../studio/instance/message/MessageService.java | 27 +++++
.../instance/message/MessageServiceTest.java | 130 +++++++++++++++++++++
2 files changed, 157 insertions(+)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
index ead1d11cc..693353e24 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/message/MessageService.java
@@ -77,6 +77,7 @@ public class MessageService {
if (!StringUtils.hasText(uniqueKey)) {
throw new BusinessException(400, "uniqueKey is required");
}
+ validateProvidedTimeWindow(startTime, endTime);
log.info("Querying message by unique key: topic={}, uniqueKey={}",
topic, uniqueKey);
return messageProvider.queryMessageByUniqueKey(instanceId, topic,
uniqueKey, startTime, endTime);
}
@@ -267,6 +268,27 @@ public class MessageService {
return StringUtils.hasText(value) ? value.trim() : null;
}
+ /**
+ * Rejects a supplied window that cannot describe a real lookup, for the
two lookups whose
+ * window is optional and therefore not covered by {@link
#validateTopicQueryWindow}: the key
+ * branch of {@code queryMessages} and {@code queryMessageByUniqueKey}.
Both used to forward a
+ * negated or inverted window to the provider, which either rejected it
with a message of its
+ * own or passed it on to the broker / cloud API, so a malformed request
had no single
+ * documented answer. No length bound is applied: the documented seven-day
maximum belongs to
+ * topic scans only.
+ */
+ private static void validateProvidedTimeWindow(Long startTime, Long
endTime) {
+ if (startTime != null && startTime < 0) {
+ throw new BusinessException(400, "message query timestamps must
not be negative");
+ }
+ if (endTime != null && endTime < 0) {
+ throw new BusinessException(400, "message query timestamps must
not be negative");
+ }
+ if (startTime != null && endTime != null && startTime >= endTime) {
+ throw new BusinessException(400, "startTime must be before
endTime");
+ }
+ }
+
private void validateTopicQueryWindow(String topic, String msgId, String
key, Long startTime, Long endTime) {
boolean hasTopic = StringUtils.hasText(topic);
boolean hasMessageId = StringUtils.hasText(msgId);
@@ -281,6 +303,11 @@ public class MessageService {
throw new BusinessException(400, "topic or msgId is required");
}
if (hasMessageId || hasKey) {
+ // A message id is a point lookup and never needs a window. A key
query may carry an
+ // explicit one, so it is validated here without the topic-scan
length bound.
+ if (hasKey) {
+ validateProvidedTimeWindow(startTime, endTime);
+ }
return;
}
long end = endTime == null ? System.currentTimeMillis() : endTime;
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
index b5921d1dd..55f0ecd17 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/message/MessageServiceTest.java
@@ -287,6 +287,78 @@ class MessageServiceTest {
verifyNoInteractions(provider, registry);
}
+ @Test
+ void keyQueryRejectsNegativeTimeWindowBeforeCallingProviderTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry,
mock(QueryHistoryService.class),
+ mock(OperationAuditService.class), ownershipGuard());
+ when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
+ when(provider.queryMessagesDetailed("instance-a", "TopicA", null,
null, "order-1", -1000L, 2000L))
+ .thenReturn(MessageQueryResult.complete(List.of()));
+
+ assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA",
null, null, "order-1",
+ -1000L, 2000L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("message query timestamps must not be negative");
+
+ verifyNoInteractions(provider, registry);
+ }
+
+ @Test
+ void keyQueryRejectsInvalidTimeWindowBeforeCallingProviderTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry,
mock(QueryHistoryService.class),
+ mock(OperationAuditService.class), ownershipGuard());
+ when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
+ when(provider.queryMessagesDetailed("instance-a", "TopicA", null,
null, "order-1", 2000L, 1000L))
+ .thenReturn(MessageQueryResult.complete(List.of()));
+ when(provider.queryMessagesDetailed("instance-a", "TopicA", null,
null, "order-1", 1000L, 1000L))
+ .thenReturn(MessageQueryResult.complete(List.of()));
+
+ assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA",
null, null, "order-1",
+ 2000L, 1000L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("startTime must be before endTime");
+ assertThatThrownBy(() -> service.queryMessages("instance-a", "TopicA",
null, null, "order-1",
+ 1000L, 1000L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("startTime must be before endTime");
+
+ verifyNoInteractions(provider, registry);
+ }
+
+ @Test
+ void keyQueryKeepsASingleSidedTimeWindowTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry,
mock(QueryHistoryService.class),
+ mock(OperationAuditService.class), ownershipGuard());
+ when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
+ when(provider.queryMessagesDetailed("instance-a", "TopicA", null,
"PAID", "order-1", 5000L, null))
+ .thenReturn(MessageQueryResult.complete(List.of()));
+
+ service.queryMessages("instance-a", "TopicA", null, "PAID", "order-1",
5000L, null);
+
+ verify(provider).queryMessagesDetailed("instance-a", "TopicA", null,
"PAID", "order-1", 5000L, null);
+ }
+
+ @Test
+ void messageIdLookupStillIgnoresTheTimeWindowTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ MessageService service = new MessageService(provider, registry,
mock(QueryHistoryService.class),
+ mock(OperationAuditService.class), ownershipGuard());
+ when(registry.byInstanceId("instance-a")).thenReturn(Optional.empty());
+ when(provider.queryMessagesDetailed("instance-a", "TopicA", "msg-001",
null, null, 2000L, 1000L))
+ .thenReturn(MessageQueryResult.complete(List.of()));
+
+ service.queryMessages("instance-a", "TopicA", "msg-001", null, null,
2000L, 1000L);
+
+ verify(provider).queryMessagesDetailed("instance-a", "TopicA",
"msg-001", null, null, 2000L, 1000L);
+ }
+
@Test
void pageQuerySlicesProviderResultAndSignalsBoundedTopicResult() {
MessageProvider provider = mock(MessageProvider.class);
@@ -458,6 +530,64 @@ class MessageServiceTest {
verifyNoInteractions(provider);
}
+ @Test
+ void uniqueKeyQueryRejectsInvertedTimeWindowBeforeCallingProviderTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ MessageService service = new MessageService(provider,
mock(InstanceProviderRegistry.class),
+ mock(QueryHistoryService.class),
mock(OperationAuditService.class), ownershipGuard());
+
+ assertThatThrownBy(() -> service.queryMessageByUniqueKey("instance-a",
"TopicA", "uniq-1",
+ 2000L, 1000L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("startTime must be before endTime");
+
+ verifyNoInteractions(provider);
+ }
+
+ @Test
+ void uniqueKeyQueryRejectsNegativeTimeWindowBeforeCallingProviderTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ MessageService service = new MessageService(provider,
mock(InstanceProviderRegistry.class),
+ mock(QueryHistoryService.class),
mock(OperationAuditService.class), ownershipGuard());
+
+ assertThatThrownBy(() -> service.queryMessageByUniqueKey("instance-a",
"TopicA", "uniq-1",
+ -1L, 1000L))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("message query timestamps must not be negative");
+
+ verifyNoInteractions(provider);
+ }
+
+ @Test
+ void uniqueKeyQueryKeepsTheTimeWindowOptionalTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ MessageService service = new MessageService(provider,
mock(InstanceProviderRegistry.class),
+ mock(QueryHistoryService.class),
mock(OperationAuditService.class), ownershipGuard());
+ MessageRecordVO record =
MessageRecordVO.builder().msgId("msg-1").build();
+ when(provider.queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", null, null))
+ .thenReturn(List.of(record));
+
+ assertThat(service.queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", null, null))
+ .containsExactly(record);
+
+ verify(provider).queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", null, null);
+ }
+
+ @Test
+ void uniqueKeyQueryAcceptsASingleSidedTimeWindowTest() {
+ MessageProvider provider = mock(MessageProvider.class);
+ MessageService service = new MessageService(provider,
mock(InstanceProviderRegistry.class),
+ mock(QueryHistoryService.class),
mock(OperationAuditService.class), ownershipGuard());
+ MessageRecordVO record =
MessageRecordVO.builder().msgId("msg-1").build();
+ when(provider.queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", 5000L, null))
+ .thenReturn(List.of(record));
+
+ assertThat(service.queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", 5000L, null))
+ .containsExactly(record);
+
+ verify(provider).queryMessageByUniqueKey("instance-a", "TopicA",
"uniq-1", 5000L, null);
+ }
+
@Test
void delegatesUniqueKeyQueryToMessageProviderTest() {
MessageProvider provider = mock(MessageProvider.class);
