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 fe44f3a47 fix(message,ai): reject empty payloads early; ignore late
stop errors (#7207)
fe44f3a47 is described below
commit fe44f3a47dd3fc8110e2617b60cf6dcb996ce784
Author: aias00 <[email protected]>
AuthorDate: Wed Oct 28 00:38:24 2026 +0800
fix(message,ai): reject empty payloads early; ignore late stop errors
(#7207)
* fix(message): reject empty payloads before resolving clients (#7207)
An empty body was only caught on the native path, and then late: after
clients had
been resolved, so a cloud redelivery of a stored message whose body came
back null
went through. Declare @NotEmpty on SendMessageDTO and add an authoritative
guard in
MetadataService.requireSendMessageRequest, the single funnel for both send
and
redelivery ahead of provider dispatch. The guard uses isEmpty() rather than
the
existing requireName/hasText helpers on purpose - a whitespace-only body
such as
" \t\n" is a legitimate payload and must reach the broker verbatim instead
of being
rejected or trimmed. The native-path check stays as defence in depth so its
FAILED
audit row is preserved, and docs/api-spec.md 5.8 records the 400.
* fix(ai): ignore stop errors after the owned run ends (#7000)
A stop request that failed after the run had already finished surfaced as
an error
and left the composer in the stopping state. Track run ownership, clear
stopRequested before the early return, and swallow the stop failure once
the run is
over.
Batch integration of two fixes by the same contributor; #7000 is closed as
merged
into this one. The review note that the AI tool path also needed the
empty-body
check did not hold on the current trunk - message.yaml already declares
body as
required with minLength 1 and ToolValidationFilter enforces it at runtime -
so the
remaining gap was the cloud redelivery path described above, now covered by
cloudRedeliveryOfAnEmptySourceBodyIsRejectedTest. Verified: 230 backend
tests
across the affected classes, 19 frontend tests for useAgentRun, prettier /
eslint /
tsc clean, and removing the guard fails both parameterized cases.
---
docs/api-spec.md | 2 +-
.../studio/instance/topic/MetadataService.java | 10 +++++
.../studio/instance/topic/SendMessageDTO.java | 2 +
.../provider/apache/RocketMQAdminClientImpl.java | 5 +++
.../studio/instance/topic/MetadataServiceTest.java | 19 ++++++++++
.../studio/instance/topic/TopicControllerTest.java | 32 ++++++++++++++++
.../apache/RocketMQAdminClientImplTest.java | 31 +++++++++++++++
web/src/pages/ai/hooks/useAgentRun.test.ts | 44 ++++++++++++++++++++++
web/src/pages/ai/hooks/useAgentRun.ts | 9 +++++
9 files changed, 153 insertions(+), 1 deletion(-)
diff --git a/docs/api-spec.md b/docs/api-spec.md
index a761e1f03..bfbfcb73c 100644
--- a/docs/api-spec.md
+++ b/docs/api-spec.md
@@ -1357,7 +1357,7 @@ POST /api/topics/send
| `topic` | `string` | 是 | Topic 名称 |
| `tag` | `string` | 否 | 消息 Tag |
| `key` | `string` | 否 | 消息 Key（用于消息查询） |
-| `body` | `string` | 是 | 消息体内容 |
+| `body` | `string` | 是 | 非空消息体；缺失、`null` 或空字符串返回 `400`，消息体按原文发送（包括空白字符） |
| `properties` | `Record<string, string>` | 否 | 消息自定义属性键值对 |
**Response `data`:**
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
index 562fa384e..715137a47 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/MetadataService.java
@@ -849,10 +849,20 @@ public class MetadataService {
return instance.getName();
}
+ /**
+ * Single funnel for every send and redelivery, ahead of the provider
dispatch, so the empty-body
+ * contract holds for cloud vendors too and for DTOs built internally (a
stored message may carry a
+ * {@code null} body) rather than only for HTTP requests that bean
validation already screens.
+ */
private void requireSendMessageRequest(SendMessageDTO request) {
if (request == null) {
throw new BusinessException(400, "Topic send message request is
required");
}
+ // isEmpty(), deliberately not requireName/hasText: a whitespace-only
body such as " \t\n" is a
+ // legitimate payload and must reach the broker verbatim instead of
being rejected or trimmed.
+ if (request.getBody() == null || request.getBody().isEmpty()) {
+ throw new BusinessException(400, "body must not be empty");
+ }
}
private static final int MAX_IMPORT_GROUPS = 100;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
index 8680495d2..1d64a4269 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/instance/topic/SendMessageDTO.java
@@ -17,6 +17,7 @@
package org.apache.rocketmq.studio.instance.topic;
import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotEmpty;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
@@ -35,6 +36,7 @@ public class SendMessageDTO {
private String topic;
private String tag;
private String key;
+ @NotEmpty(message = "body must not be empty")
private String body;
private Map<String, String> properties;
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index 51e63c093..97f04ae8b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -678,6 +678,11 @@ public class RocketMQAdminClientImpl implements
AdminClient {
private void validateMessageSize(SendMessageDTO request) {
int size = request.getBody() == null ? 0 :
request.getBody().getBytes(StandardCharsets.UTF_8).length;
+ if (size == 0) {
+ String message = "Message body must not be empty";
+ recordAudit("SEND_MESSAGE", request.getTopic(), message, "FAILED");
+ throw new BusinessException(400, message);
+ }
if (size > MAX_MESSAGE_SIZE) {
String message = "Message body size " + size
+ " exceeds the maximum of " + MAX_MESSAGE_SIZE + " bytes";
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
index 831c3796e..14dcef254 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/MetadataServiceTest.java
@@ -52,6 +52,7 @@ import
org.apache.rocketmq.studio.provider.apache.MetadataProvider;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.junit.jupiter.params.provider.ValueSource;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -182,6 +183,24 @@ class MetadataServiceTest {
assertThat(request.getValue().getProperties()).containsOnlyKeys("tenant");
}
+ @ParameterizedTest
+ @NullAndEmptySource
+ void cloudRedeliveryOfAnEmptySourceBodyIsRejectedTest(String sourceBody) {
+ // Redelivery builds its DTO internally, so the @NotEmpty screening on
the HTTP send request
+ // never sees it, and a stored message legitimately reports a null
body. Without the guard in
+ // requireSendMessageRequest a cloud instance would forward the empty
payload to the vendor
+ // SDK and surface its failure instead of a local 400.
+ MessageRecordVO source =
MessageRecordVO.builder().topic("orders").body(sourceBody)
+ .properties(Map.of("tenant", "alpha")).build();
+
+ assertThatThrownBy(() ->
metadataService.redeliverMessage("cloud-instance", "group-a", source, null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("body must not be empty")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+ verifyNoInteractions(cloudProvider);
+ }
+
@Test
void redeliverMessageShouldFilterSystemReservedPropertiesTest() {
// §15.5.1: copying KEYS/TAGS/UNIQ_KEY verbatim makes putUserProperty
reject the send.
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
index 702b37884..90e2f4272 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/instance/topic/TopicControllerTest.java
@@ -26,6 +26,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.rocketmq.studio.common.domain.enums.TopicPerm;
import org.apache.rocketmq.studio.common.domain.enums.TopicType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc;
@@ -386,6 +388,36 @@ class TopicControllerTest extends WebMvcAuthTestSupport {
verifyNoInteractions(metadataService);
}
+ @ParameterizedTest
+ @ValueSource(strings = {
+ "{\"instanceId\":\"instance-a\",\"topic\":\"orders\"}",
+ "{\"instanceId\":\"instance-a\",\"topic\":\"orders\",\"body\":null}",
+ "{\"instanceId\":\"instance-a\",\"topic\":\"orders\",\"body\":\"\"}"
+ })
+ void sendRejectsMissingOrEmptyBodyBeforeCallingMetadataTest(String json)
throws Exception {
+
mockMvc.perform(post("/api/topics/send").contentType(MediaType.APPLICATION_JSON).content(json))
+ .andExpect(status().isBadRequest())
+ .andExpect(jsonPath("$.code").value(400))
+ .andExpect(jsonPath("$.message").value("body must not be
empty"));
+
+ verifyNoInteractions(metadataService);
+ }
+
+ @Test
+ void sendPreservesWhitespaceOnlyBodyTest() throws Exception {
+ String body = " \t\n";
+ SendMessageDTO request =
SendMessageDTO.builder().instanceId("instance-a").topic("orders")
+ .body(body).build();
+
+
mockMvc.perform(post("/api/topics/send").contentType(MediaType.APPLICATION_JSON)
+ .content(objectMapper.writeValueAsString(request)))
+ .andExpect(status().isOk());
+
+ ArgumentCaptor<SendMessageDTO> captor =
ArgumentCaptor.forClass(SendMessageDTO.class);
+ verify(metadataService).sendMessage(captor.capture());
+ assertThat(captor.getValue().getBody()).isEqualTo(body);
+ }
+
@Test
void updateTopicBindsLiteTtlMinutesSoTheServiceCanRejectItTest() throws
Exception {
TopicVO updated = new TopicVO();
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index 03a54214c..6f7f710c8 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -2101,6 +2101,37 @@ class RocketMQAdminClientImplTest {
assertThat(result.getMsgId()).isEqualTo("msg-1");
}
+ @ParameterizedTest
+ @NullAndEmptySource
+ void emptyBodyIsRejectedBeforeOwnershipOrClientResolutionTest(String body)
{
+ SendMessageDTO request =
SendMessageDTO.builder().instanceId("instance-a").topic("TopicA")
+ .body(body).build();
+
+ assertThatThrownBy(() -> adminClient.sendMessage(request))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Message body must not be empty")
+ .satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(400));
+
+ verifyNoInteractions(ownershipGuard, runtimeAdminClientResolver,
adminExt, sendProducer, clientPool);
+ }
+
+ @Test
+ void whitespaceBodyIsSentWithoutTrimmingTest() throws Exception {
+ SendResult sent = new SendResult();
+ sent.setSendStatus(SendStatus.SEND_OK);
+ sent.setMsgId("whitespace-message");
+ when(sendProducer.send(any(Message.class))).thenReturn(sent);
+ String body = " \t\n";
+ SendMessageDTO request =
SendMessageDTO.builder().instanceId("instance-a").topic("TopicA")
+ .body(body).build();
+
+
assertThat(adminClient.sendMessage(request).getMsgId()).isEqualTo("whitespace-message");
+
+ ArgumentCaptor<Message> captor =
ArgumentCaptor.forClass(Message.class);
+ verify(sendProducer).send(captor.capture());
+
assertThat(captor.getValue().getBody()).isEqualTo(body.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+ }
+
@Test
void
sendMessageShouldRejectOversizedUtf8BodyBeforeResolvingEndpointOrCreatingProducer()
{
SendMessageDTO request = new SendMessageDTO();
diff --git a/web/src/pages/ai/hooks/useAgentRun.test.ts
b/web/src/pages/ai/hooks/useAgentRun.test.ts
index b742f98f6..4f7ece3ea 100644
--- a/web/src/pages/ai/hooks/useAgentRun.test.ts
+++ b/web/src/pages/ai/hooks/useAgentRun.test.ts
@@ -394,6 +394,50 @@ describe('useAgentRun', () => {
expect(result.current.canStop).toBe(false);
});
+ it.each(['terminal frame', 'closed stream', 'failed stream'] as const)(
+ 'ignoresAStopRejectionAfterTheOwnedRunEndsTest (%s)',
+ async (ending) => {
+ const pendingStop = deferred<Awaited<ReturnType<typeof stopRun>>>();
+ vi.mocked(stopRun).mockReturnValue(pendingStop.promise);
+ const onError = vi.fn();
+ const { result } = render({ onError });
+ let sent!: Promise<void>;
+ let stopped!: Promise<void>;
+ await act(async () => {
+ sent = result.current.send(7, { message: 'hi' });
+ openedStreams[0].emit(runStarted());
+ stopped = result.current.stop();
+ });
+ await act(async () => {
+ if (ending === 'terminal frame') openedStreams[0].emit(runFinished());
+ else {
+ if (ending === 'failed stream') openedStreams[0].fail(new
Error('stream failed'));
+ else openedStreams[0].finish();
+ await sent;
+ }
+ });
+ const previousError = result.current.error;
+ const previousErrorCalls = onError.mock.calls.length;
+ await act(async () => {
+ pendingStop.reject(new Error('late stop failed'));
+ await stopped;
+ });
+ expect(result.current.error).toBe(previousError);
+ expect(onError).toHaveBeenCalledTimes(previousErrorCalls);
+ // Asserted while the stream is still open in the `terminal frame` case:
swallowing the
+ // rejection must leave the button back on `stop`, not stuck on
`stopping`.
+ expect(result.current.stopRequested).toBe(false);
+ if (ending === 'terminal frame') {
+ expect(result.current.lastStatus).toBe('COMPLETED');
+ await act(async () => {
+ openedStreams[0].finish();
+ await sent;
+ });
+ }
+ expect(result.current.stopRequested).toBe(false);
+ },
+ );
+
it('reportsAStaleStopRejectionAndLeavesTheStreamOpenTest', async () => {
const onError = vi.fn();
vi.mocked(stopRun).mockRejectedValue(new Error('该会话已有正在进行的回答'));
diff --git a/web/src/pages/ai/hooks/useAgentRun.ts
b/web/src/pages/ai/hooks/useAgentRun.ts
index 4ee5fcf59..01e205643 100644
--- a/web/src/pages/ai/hooks/useAgentRun.ts
+++ b/web/src/pages/ai/hooks/useAgentRun.ts
@@ -168,6 +168,7 @@ export function useAgentRun(
const generationRef = useRef(0);
const streamRequestIdRef = useRef(0);
const chatInFlightRef = useRef(false);
+ const runEndedRef = useRef(true);
const abortControllerRef = useRef<AbortController | null>(null);
const frameRef = useRef<number | null>(null);
const runIdRef = useRef<number | null>(null);
@@ -211,6 +212,7 @@ export function useAgentRun(
optionsRef.current.onRunStarted?.(event);
break;
case 'run_finished':
+ runEndedRef.current = true;
setLastStatus(event.status);
optionsRef.current.onRunFinished?.(event);
break;
@@ -253,6 +255,7 @@ export function useAgentRun(
if (abortControllerRef.current === controller)
abortControllerRef.current = null;
// A newer stream (or a navigation) owns the guard and UI now.
if (requestId !== streamRequestIdRef.current) return;
+ runEndedRef.current = true;
// Released before the refetch is awaited on purpose: a reload that
hangs must not leave the
// composer unable to send. A send that overtakes this finally block
bumps the request id, and
// the guard after the await below then keeps this run's cleanup off the
newer one's state.
@@ -317,6 +320,7 @@ export function useAgentRun(
return;
}
chatInFlightRef.current = true;
+ runEndedRef.current = false;
// Only `send` carries a prompt; a re-attach finds the user row already
in the timeline.
if (userMessage !== undefined) setPendingUserMessage(userMessage);
@@ -406,7 +410,12 @@ export function useAgentRun(
// Navigation or a newer stream invalidated this Stop request. Its
failure belongs to the old
// run and must not paint an error onto the conversation that owns the
hook now.
if (requestId !== streamRequestIdRef.current || generation !==
generationRef.current) return;
+ // Reset before the ownership guard below: the run that owns this Stop
may have ended while the
+ // stream is still open, and nothing else clears `stopping` until the
stream settles.
setStopRequested(false);
+ // The run already reached a terminal state, so a Stop rejection is
expected; surfacing it would
+ // paint a failure for a run that is over onto the live conversation.
+ if (runEndedRef.current) return;
setError(describeThrownMessage(stopError));
optionsRef.current.onError?.(stopError);
}
