From 19b0c49db4f9fb11a56b706ea397d5d7638819dc Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Wed, 9 Sep 2026 14:35:50 +0800 Subject: [PATCH] fix(rocketmq): handle query empty responses Closes #8469 --- agents/drivers/rocketmq/messages.go | 29 ++++--- agents/drivers/rocketmq/messages_test.go | 104 +++++++++++++++++++++++ 2 files changed, 121 insertions(+), 12 deletions(-) diff --git a/agents/drivers/rocketmq/messages.go b/agents/drivers/rocketmq/messages.go index 3872237c5e..c3854365d5 100644 --- a/agents/drivers/rocketmq/messages.go +++ b/agents/drivers/rocketmq/messages.go @@ -16,7 +16,7 @@ import ( ) const sendMessageRequestCode = 10 -const queryMessageNotFoundCode = 208 +const queryMessageNotFoundCode = 22 type messageQueueTarget struct { BrokerName string @@ -208,9 +208,6 @@ func (a *rocketMQAgent) queryMessageByKey(params map[string]any) (any, error) { messages, err := queryMessagesByKey(ctx, client, config.ConnectTimeout, topic, key, maxNum, int64Value(params, 0, "begin"), int64Value(params, time.Now().UnixMilli(), "end")) if err != nil { - if isEmptyMessageQueryError(err) { - return emptyMessageQuery(), nil - } return nil, err } return messageQueryResult(topic, messages), nil @@ -296,11 +293,14 @@ func (a *rocketMQAgent) queryMessageByTopic(params map[string]any) (any, error) ctx, cancel := context.WithTimeout(context.Background(), config.RequestTimeout) defer cancel() maxNum := min(max(1, intValue(params, 32, "maxNum")), 200) + beginTimestamp := int64Value(params, 0, "begin") + endTimestamp := int64Value(params, time.Now().UnixMilli(), "end") messages, err := client.QueryMessageByTime(ctx, topic, - int64Value(params, 0, "begin"), int64Value(params, time.Now().UnixMilli(), "end"), maxNum) + beginTimestamp, endTimestamp, maxNum) if err != nil { return nil, err } + messages = filterMessagesByStoreTimestamp(messages, beginTimestamp, endTimestamp) return messageQueryResult(topic, messages), nil } @@ -491,10 +491,6 @@ func messageQueryResult(topic string, messages []*admin.MessageExt) map[string]a return map[string]any{"messages": rows, "indexLastUpdateTimestamp": int64(0)} } -func emptyMessageQuery() map[string]any { - return map[string]any{"messages": []map[string]any{}, "indexLastUpdateTimestamp": int64(0)} -} - func sortMessageRows(messages []map[string]any) { sort.Slice(messages, func(i, j int) bool { leftTime := anyInt64(messages[i]["timestamp"]) @@ -556,9 +552,18 @@ func anyInt64(value any) int64 { } } -func isEmptyMessageQueryError(err error) bool { - message := strings.ToLower(err.Error()) - return strings.Contains(message, "208") || strings.Contains(message, "no message") || strings.Contains(message, "未找到消息") +func filterMessagesByStoreTimestamp(messages []*admin.MessageExt, beginTimestamp, endTimestamp int64) []*admin.MessageExt { + filtered := make([]*admin.MessageExt, 0, len(messages)) + for _, message := range messages { + if beginTimestamp > 0 && message.StoreTimestamp < beginTimestamp { + continue + } + if endTimestamp > 0 && message.StoreTimestamp > endTimestamp { + continue + } + filtered = append(filtered, message) + } + return filtered } func isReservedMessageProperty(key string) bool { diff --git a/agents/drivers/rocketmq/messages_test.go b/agents/drivers/rocketmq/messages_test.go index a354f7efbb..334fdd67f2 100644 --- a/agents/drivers/rocketmq/messages_test.go +++ b/agents/drivers/rocketmq/messages_test.go @@ -54,6 +54,110 @@ func TestBuildQueryMessageCommandUsesBrokerHeaderNames(t *testing.T) { } } +func TestInvokeRemotingAllowCodesAcceptsQueryMessageNotFound(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + done := make(chan error, 1) + go func() { + connection, acceptErr := listener.Accept() + if acceptErr != nil { + done <- acceptErr + return + } + defer connection.Close() + frame, readErr := readRemotingFrame(connection) + if readErr != nil { + done <- readErr + return + } + request, decodeErr := remoting.Decode(frame[4:]) + if decodeErr != nil { + done <- decodeErr + return + } + response := &remoting.RemotingCommand{Code: queryMessageNotFoundCode, Opaque: request.Opaque, Flag: 1} + encoded, encodeErr := response.Encode() + if encodeErr == nil { + _, encodeErr = connection.Write(encoded) + } + done <- encodeErr + }() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := invokeRemotingAllowCodes(ctx, listener.Addr().String(), time.Second, + remoting.NewRequest(remoting.QueryMessage, nil), remoting.Success, queryMessageNotFoundCode); err != nil { + t.Fatal(err) + } + if err := <-done; err != nil { + t.Fatal(err) + } +} + +func TestInvokeRemotingAllowCodesPreservesOtherErrors(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + done := make(chan error, 1) + go func() { + connection, acceptErr := listener.Accept() + if acceptErr != nil { + done <- acceptErr + return + } + defer connection.Close() + frame, readErr := readRemotingFrame(connection) + if readErr != nil { + done <- readErr + return + } + request, decodeErr := remoting.Decode(frame[4:]) + if decodeErr != nil { + done <- decodeErr + return + } + response := &remoting.RemotingCommand{Code: remoting.ConsumerNotOnline, Opaque: request.Opaque, Flag: 1} + encoded, encodeErr := response.Encode() + if encodeErr == nil { + _, encodeErr = connection.Write(encoded) + } + done <- encodeErr + }() + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, invokeErr := invokeRemotingAllowCodes(ctx, listener.Addr().String(), time.Second, + remoting.NewRequest(remoting.QueryMessage, nil), remoting.Success, queryMessageNotFoundCode) + if invokeErr == nil || !strings.Contains(invokeErr.Error(), "206") { + t.Fatalf("expected ConsumerNotOnline to remain an error, got %v", invokeErr) + } + if err := <-done; err != nil { + t.Fatal(err) + } +} + +func TestFilterMessagesByStoreTimestampUsesInclusiveBounds(t *testing.T) { + messages := []*admin.MessageExt{ + {MsgId: "before", StoreTimestamp: 99}, + {MsgId: "begin", StoreTimestamp: 100}, + {MsgId: "middle", StoreTimestamp: 150}, + {MsgId: "end", StoreTimestamp: 200}, + {MsgId: "after", StoreTimestamp: 201}, + } + filtered := filterMessagesByStoreTimestamp(messages, 100, 200) + if len(filtered) != 3 { + t.Fatalf("expected three messages in range, got %d", len(filtered)) + } + for index, messageID := range []string{"begin", "middle", "end"} { + if filtered[index].MsgId != messageID { + t.Fatalf("unexpected message at index %d: %s", index, filtered[index].MsgId) + } + } +} + func TestInvokeRemotingAllowCodes(t *testing.T) { listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil {