You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by xi...@apache.org on 2026/10/10 19:00:37 UTC
(eventmesh-dashboard) branch dev-maven-release updated: feat: 采集 RocketMQ (#232)
This is an automated email from the ASF dual-hosted git repository.
Pil0tXia pushed a commit to branch dev-maven-release
in repository https://gitbox.apache.org/repos/asf/eventmesh-dashboard.git
The following commit(s) were added to refs/heads/dev-maven-release by this push:
new d1a81761 feat: 采集 RocketMQ (#232)
d1a81761 is described below
commit d1a817619d3da9d717e6c14a4e2ba9bd922e7319
Author: SUPERSAIYAN <17...@users.noreply.github.com>
AuthorDate: Sun Oct 11 03:00:30 2026 +0800
feat: 采集 RocketMQ (#232)
* 实现 RocketMQ Topic 和消费位点异步采集
* 抽象成一个采集topic方法
* 重构 RocketMQ 采集流程，拆分回调并补充中文注释
- 提取单 Topic 采集入口，统一组织位点和消费组采集
- 使用具名内部类封装各类响应处理
* 新增2个模型类
* 重构 RocketMQ 采集流程并完善独立指标模型
* 改为使用计数器来控制调用开始和结束
* 增加Licensed
* 新增采集
Broker 消息统计 写入、取出消息的分钟／小时／日窗口计数，以及对应 TPS
Broker 运行状态 分发积压字节数、刷盘积压字节数、最早保留消息距今时间、各线程池排队数量
* 拆分采集轮次管理与 Topic、Broker 采集职责
* 补充 RocketMQ 消费状态、磁盘和消息统计采集
* 补充 RocketMQ 累计消息指标采集
* 新增单测
* 失败的时候写为0，不再跳过
* 失败的时候写为0，不再跳过，使用3位运算符
* 把null 设置为 0 抽象为一个方法
* test: 新增带生产者消费者的采集手动测试，验证消费侧指标采集
* 修复采集 PR 的依赖版本和许可证冲突
* 修复采集缓存并发写入丢失样本
* 采集写入前检查关闭状态并补充测试
---------
Co-authored-by: sifu <14...@qq.com>
---
eventmesh-dashboard-console/pom.xml | 55 +
.../function/report/ReportHandlerManage.java | 1 +
.../ReportTag.java} | 22 +-
.../function/report/collect/CollectManage.java | 1 +
.../function/report/collect/RestoreData.java | 7 +-
.../report/collect/exporter/RocketMQCollect.java | 724 ++++++++
.../report/iotdb/IotDBReportMetaHandler.java | 3 +-
.../model/rocketmq/Rocketmq2ProducerOffset.java | 16 +-
...sInTotal.java => RocketmqBrokerMessagesIn.java} | 18 +-
...InTotal.java => RocketmqBrokerMessagesOut.java} | 18 +-
....java => RocketmqConsumerConnectionNumber.java} | 16 +-
...InTotal.java => RocketmqConsumerFailedTps.java} | 21 +-
.../rocketmq/RocketmqConsumerGroupNumber.java | 4 +-
.../model/rocketmq/RocketmqConsumerLagLatency.java | 22 +-
...gesInTotal.java => RocketmqConsumerOffset.java} | 26 +-
...Total.java => RocketmqConsumerProcessTime.java} | 21 +-
...Number.java => RocketmqConsumerSuccessTps.java} | 22 +-
...sInTotal.java => RocketmqGroupMessagesOut.java} | 24 +-
.../model/rocketmq/RocketmqMessagesInTotal.java | 12 +-
.../model/rocketmq/RocketmqMessagesOutTotal.java | 12 +-
...mber.java => RocketmqStorageDiskFreeBytes.java} | 10 +-
...sInTotal.java => RocketmqStorageDiskUsage.java} | 18 +-
.../RocketmqStorageMessageReserveTime.java | 6 +-
.../rocketmq/RocketmqThreadPoolWartermark.java | 11 +-
.../model/rocketmq/RocketmqThroughputInTotal.java | 6 +-
...esInTotal.java => RocketmqTopicMessagesIn.java} | 21 +-
.../report/model/rocketmq/RocketmqTopicNumber.java | 2 +-
.../mapstruct/report/RocketMQCollectMapper.java | 194 ++
.../report/collect/DataSyncHandlerTest.java | 156 ++
.../report/collect/RocketMQCollectManualTest.java | 123 ++
.../RocketMQCollectWithClientManualTest.java | 229 +++
.../collect/exporter/RocketMQCollectTest.java | 1891 ++++++++++++++++++++
32 files changed, 3590 insertions(+), 122 deletions(-)
diff --git a/eventmesh-dashboard-console/pom.xml b/eventmesh-dashboard-console/pom.xml
index 59513177..967fbbf3 100644
--- a/eventmesh-dashboard-console/pom.xml
+++ b/eventmesh-dashboard-console/pom.xml
@@ -243,4 +243,59 @@
</plugin>
</plugins>
</build>
+ <profiles>
+ <profile>
+ <id>rocketmq-collect-tests</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <testIncludes combine.self="override">
+ <testInclude>**/RocketMQCollectTest.java</testInclude>
+ <testInclude>**/DataSyncHandlerTest.java</testInclude>
+ </testIncludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <testFailureIgnore>false</testFailureIgnore>
+ <includes>
+ <include>**/RocketMQCollectTest.java</include>
+ <include>**/DataSyncHandlerTest.java</include>
+ </includes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>rocketmq-manual-collect</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <testIncludes combine.self="override">
+ <testInclude>**/RocketMQCollectManualTest.java</testInclude>
+ </testIncludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <includes>
+ <include>**/RocketMQCollectManualTest.java</include>
+ </includes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
</project>
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/ReportHandlerManage.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/ReportHandlerManage.java
index a8516bfe..a97404d8 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/ReportHandlerManage.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/ReportHandlerManage.java
@@ -117,6 +117,7 @@ public class ReportHandlerManage {
@Getter
private final Map<String, ReportMetaData> aggregationMetaDataMap = new HashMap<>();
@Setter
+ @Getter
private ReportEngine reportEngine;
@Setter
private ReportConfig reportConfig;
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/annotation/ReportTag.java
similarity index 56%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/annotation/ReportTag.java
index c566a743..c1d07640 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/annotation/ReportTag.java
@@ -15,19 +15,15 @@
* limitations under the License.
*/
-package org.apache.eventmesh.dashboard.console.function.report.collect.exporter;
+package org.apache.eventmesh.dashboard.console.function.report.annotation;
-import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
-import org.apache.eventmesh.dashboard.common.enums.ClusterType;
-import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
-
-@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
- ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
-public class RocketMQCollect extends AbstractCollect {
-
- @Override
- protected void doCollect() {
-
- }
+/** 显式声明报表维度，允许资源名称作为标签，无需将名称伪装成 ID。 */
+@Target(ElementType.FIELD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface ReportTag {
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/CollectManage.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/CollectManage.java
index f3b7369d..fe5fb5fd 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/CollectManage.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/CollectManage.java
@@ -164,6 +164,7 @@ public class CollectManage {
public void init() {
this.dataSyncHandler.setPaddingService(paddingService);
+ this.dataSyncHandler.setReportEngine(this.reportHandlerManage.getReportEngine());
}
public void collect() {
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/RestoreData.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/RestoreData.java
index ee046618..366ec966 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/RestoreData.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/RestoreData.java
@@ -20,9 +20,10 @@ package org.apache.eventmesh.dashboard.console.function.report.collect;
import org.apache.eventmesh.dashboard.common.model.metadata.CollectMetadata;
import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import lombok.Getter;
import lombok.Setter;
@@ -42,12 +43,12 @@ public class RestoreData {
private int index;
@Getter
- private Map<Class<?>, List<Object>> dataMap = new HashMap<>();
+ private Map<Class<?>, List<Object>> dataMap = new ConcurrentHashMap<>();
public void setData(Object data) {
Class<?> clazz = data.getClass();
- dataMap.computeIfAbsent(clazz, k -> new ArrayList<>()).add(data);
+ dataMap.computeIfAbsent(clazz, k -> Collections.synchronizedList(new ArrayList<>())).add(data);
}
public void restore() {
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java
index c566a743..c87ff860 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollect.java
@@ -19,15 +19,739 @@ package org.apache.eventmesh.dashboard.console.function.report.collect.exporter;
import org.apache.eventmesh.dashboard.common.annotation.ClusterTypeMark;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
import org.apache.eventmesh.dashboard.console.function.report.collect.AbstractCollect;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.OrganizationId;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.header.GetAllSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryConsumeTimeSpanRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryTopicConsumeByWhoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+
+import java.io.StringReader;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import lombok.Setter;
@ClusterTypeMark(clusterType = {ClusterType.STORAGE_ROCKETMQ_CLUSTER, ClusterType.STORAGE_ROCKETMQ_BROKER,
ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE, ClusterType.STORAGE_ROCKETMQ_BROKER_RAFT})
public class RocketMQCollect extends AbstractCollect {
+ private static final RocketMQCollectMapper MAPPER = RocketMQCollectMapper.INSTANCE;
+ private static final Pattern BYTE_COUNT = Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|[KMGTPE]iB)");
+
+ private static final Pattern DISK_CAPACITY = Pattern.compile("Total : (.+), Free : (.+)\\.");
+
+ private String runtimeUnique;
+
+ @Setter
+ private DefaultRemotingClient defaultRemotingClient;
+
+ @Override
+ public void setRuntimeMetadata(RuntimeMetadata runtimeMetadata) {
+ super.setRuntimeMetadata(runtimeMetadata);
+ this.runtimeUnique = runtimeMetadata == null ? null : runtimeMetadata.getUnique();
+ }
@Override
protected void doCollect() {
+ if (this.runtimeUnique == null) {
+ return;
+ }
+ this.defaultRemotingClient = SDKManage.getInstance().getClient(SDKTypeEnum.ADMIN, this.runtimeUnique);
+ new CollectRound().executeAndAwait();
+ }
+
+ /**
+ * 一轮采集流程：
+ * 1. 先注册 Topic 和 Broker 两个方向的根请求，再分别启动，统一管理请求计数。
+ * 2. TopicCollect 采集 Topic 总数、位点、积压时间、消息统计及客户端处理指标；死信 Topic 不查询消费组。
+ * 3. BrokerCollect 采集配置消费组总数、消息统计和运行状态；响应直接调用父类 setData。
+ * 4. 等待所有异步请求完成后返回，父类再统一交付本轮数据。
+ * 5. 超时或中断时保留已经采集的数据，关闭本轮，拒绝尚未开始处理的迟到回调。
+ */
+ private class CollectRound {
+ private final DefaultRemotingClient client = defaultRemotingClient;
+ private final AtomicInteger pending = new AtomicInteger();
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final long deadline = System.currentTimeMillis() + 4000;
+ private volatile boolean closed;
+
+ private void executeAndAwait() {
+ try {
+ // 先注册所有根请求，避免同步回调在后续请求注册前让计数归零。
+ TopicCollect topics = new TopicCollect();
+ BrokerCollect broker = new BrokerCollect();
+ topics.start();
+ broker.start();
+ // 所有请求完成时计数归零并唤醒；超时则保留已采集的数据。
+ this.completed.await(Math.max(1, this.deadline - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ this.close();
+ }
+ }
+
+ private void close() {
+ this.closed = true;
+ }
+
+ /** 每次追加样本前检查本轮状态，已关闭则放弃本次写入。 */
+ private void setData(OrganizationId data) {
+ if (this.closed) {
+ return;
+ }
+ RocketMQCollect.this.setData(data);
+ }
+
+ /** 采集 Topic 总数、生产和消费位点、积压量及时间、消息统计、消费组连接数和客户端处理指标。 */
+ private class TopicCollect {
+ private final Set<String> connectedGroups = ConcurrentHashMap.newKeySet();
+ private final TopicsCallback topics =
+ new TopicsCallback(RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null));
+
+ private void start() {
+ this.topics.execute();
+ }
+
+ private void collectTopic(String topic) {
+ this.collectTopicStats(topic);
+ this.collectMessages(topic, null);
+ if (!topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
+ this.collectGroups(topic);
+ }
+ }
+
+ /** 原生统计键：写入按 Topic，取出按 Topic@消费组。 */
+ private void collectMessages(String topic, String group) {
+ ViewBrokerStatsDataRequestHeader header = new ViewBrokerStatsDataRequestHeader();
+ header.setStatsName(group == null ? "TOPIC_PUT_NUMS" : "GROUP_GET_NUMS");
+ header.setStatsKey(group == null ? topic : topic + "@" + group);
+ new MessageStatsCallback(RemotingCommand.createRequestCommand(RequestCode.VIEW_BROKER_STATS_DATA, header),
+ topic, group).execute();
+ }
+
+ private void collectTopicStats(String topic) {
+ GetTopicStatsInfoRequestHeader topicHeader = new GetTopicStatsInfoRequestHeader();
+ topicHeader.setTopic(topic);
+ RemotingCommand topicRequest = RemotingCommand.createRequestCommand(RequestCode.GET_TOPIC_STATS_INFO, topicHeader);
+ new TopicStatsCallback(topicRequest, topic).execute();
+ }
+
+ private void collectGroups(String topic) {
+ QueryTopicConsumeByWhoRequestHeader groupHeader = new QueryTopicConsumeByWhoRequestHeader();
+ groupHeader.setTopic(topic);
+ RemotingCommand groupRequest = RemotingCommand.createRequestCommand(RequestCode.QUERY_TOPIC_CONSUME_BY_WHO, groupHeader);
+ new GroupsCallback(groupRequest, topic).execute();
+ }
+
+ private void collectConnections(String group) {
+ GetConsumerConnectionListRequestHeader connectionHeader = new GetConsumerConnectionListRequestHeader();
+ connectionHeader.setConsumerGroup(group);
+ RemotingCommand connectionRequest = RemotingCommand.createRequestCommand(
+ RequestCode.GET_CONSUMER_CONNECTION_LIST, connectionHeader);
+ new ConnectionsCallback(connectionRequest, group).execute();
+ }
+
+ private void collectConsumeStats(String topic, String group) {
+ GetConsumeStatsRequestHeader consumeHeader = new GetConsumeStatsRequestHeader();
+ consumeHeader.setTopic(topic);
+ consumeHeader.setConsumerGroup(group);
+ RemotingCommand consumeRequest = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUME_STATS, consumeHeader);
+ new ConsumeStatsCallback(consumeRequest, topic, group).execute();
+ }
+
+ /** 步骤一：解析 Broker 的 Topic 列表，逐个启动采集。 */
+ private class TopicsCallback extends CollectCallback {
+ private TopicsCallback(RemotingCommand request) {
+ super(request);
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ TopicConfigSerializeWrapper topics =
+ RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), TopicConfigSerializeWrapper.class);
+ // 缺少列表不是零；仅对成功返回的完整配置表计数。
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("topicConfigTable") instanceof Map<?, ?>)) {
+ return;
+ }
+ CollectRound.this.setData(MAPPER.topicNumber((long) topics.getTopicConfigTable().size()));
+ for (String topic : topics.getTopicConfigTable().keySet()) {
+ collectTopic(topic);
+ }
+ }
+ }
+
+ private class MessageStatsCallback extends WindowStatsCallback {
+ private final String topic;
+ private final String group;
+
+ private MessageStatsCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void recordWindow(String window, long count, float rate) {
+ CollectRound.this.setData(this.group == null
+ ? MAPPER.topicMessages(this.topic, window, count, rate)
+ : MAPPER.groupMessages(this.topic, this.group, window, count, rate));
+ }
+ }
+
+ /** 步骤二：采集队列位点，并计算 Topic 汇总位点。 */
+ private class TopicStatsCallback extends CollectCallback {
+ private final String topic;
+
+ private TopicStatsCallback(RemotingCommand request, String topic) {
+ super(request);
+ this.topic = topic;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ TopicStatsTable stats =
+ RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), TopicStatsTable.class);
+ stats.getOffsetTable().forEach((queue, offset) -> {
+ String queueId = String.valueOf(queue.getQueueId());
+
+ CollectRound.this.setData(MAPPER.topicOffset(topic, queueId, offset));
+ });
+ long min = stats.getOffsetTable().values().stream().mapToLong(offset -> offset.getMinOffset()).sum();
+ long max = stats.getOffsetTable().values().stream().mapToLong(offset -> offset.getMaxOffset()).sum();
+ long lastUpdate = stats.getOffsetTable().values().stream()
+ .mapToLong(offset -> offset.getLastUpdateTimestamp()).max().orElse(0L);
+
+ CollectRound.this.setData(MAPPER.aggregateOffset(topic, min, max, lastUpdate));
+ }
+ }
+
+ /** 步骤三：发现 Topic 对应的消费组，启动连接与消费进度采集。 */
+ private class GroupsCallback extends CollectCallback {
+ private final String topic;
+
+ private GroupsCallback(RemotingCommand request, String topic) {
+ super(request);
+ this.topic = topic;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ GroupList groups = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), GroupList.class);
+ for (String group : groups.getGroupList()) {
+ if (TopicCollect.this.connectedGroups.add(group)) {
+ collectConnections(group);
+ }
+ collectConsumeStats(topic, group);
+ collectMessages(topic, group);
+ }
+ }
+ }
+
+ /** 步骤四：采集消费组连接数；同一消费组每轮只查询一次。 */
+ private class ConnectionsCallback extends CollectCallback {
+ private final String group;
+
+ private ConnectionsCallback(RemotingCommand request, String group) {
+ super(request);
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ ConsumerConnection connections = RemotingSerializable.decode(
+ responseFuture.getResponseCommand().getBody(), ConsumerConnection.class);
+
+ CollectRound.this.setData(MAPPER.connections(group, connections.getConnectionSet().size()));
+ // 每个在线客户端分别采样，不把 Broker 取消息 TPS 当成处理成功 TPS。
+ Set<String> clients = ConcurrentHashMap.newKeySet();
+ connections.getConnectionSet().forEach(connection -> {
+ String clientId = connection.getClientId();
+ if (clientId != null && !clientId.isBlank() && clients.add(clientId)) {
+ GetConsumerRunningInfoRequestHeader header = new GetConsumerRunningInfoRequestHeader();
+ header.setConsumerGroup(this.group);
+ header.setClientId(clientId);
+ header.setJstackEnable(false);
+ new ClientStatsCallback(RemotingCommand.createRequestCommand(
+ RequestCode.GET_CONSUMER_RUNNING_INFO, header), this.group, clientId).execute();
+ }
+ });
+ }
+ }
+
+ /** 按 Topic、消费组和客户端采集消费成功 TPS、失败 TPS 和平均处理耗时。 */
+ private class ClientStatsCallback extends CollectCallback {
+ private final String group;
+ private final String clientId;
+
+ private ClientStatsCallback(RemotingCommand request, String group, String clientId) {
+ super(request);
+ this.group = group;
+ this.clientId = clientId;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("statusTable") instanceof Map<?, ?> statuses)) {
+ return;
+ }
+ statuses.forEach((key, value) -> {
+ if (!(key instanceof String topic) || !(value instanceof Map<?, ?> status)) {
+ return;
+ }
+ Float success = nonnegativeFloatOrZero(status.get("consumeOKTPS"));
+ Float failed = nonnegativeFloatOrZero(status.get("consumeFailedTPS"));
+ Float elapsed = nonnegativeFloatOrZero(status.get("consumeRT"));
+ CollectRound.this.setData(MAPPER.consumerSuccess(topic, this.group, this.clientId, success));
+ CollectRound.this.setData(MAPPER.consumerFailed(topic, this.group, this.clientId, failed));
+ CollectRound.this.setData(MAPPER.consumerProcessTime(topic, this.group, this.clientId, elapsed));
+ });
+ }
+ }
+
+ /** 采集各 Topic、消费组和队列中最早未消费消息的积压时间。 */
+ private class LagCallback extends CollectCallback {
+ private final String topic;
+ private final String group;
+
+ private LagCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body == null || !(body.get("consumeTimeSpanSet") instanceof List<?> spans)) {
+ return;
+ }
+ for (Object item : spans) {
+ if (!(item instanceof Map<?, ?> span) || !(span.get("messageQueue") instanceof Map<?, ?> queue)
+ || !this.topic.equals(queue.get("topic"))) {
+ continue;
+ }
+ Long queueId = nonnegativeLong(queue.get("queueId"));
+ Long delay = nonnegativeLong(span.get("delayTime"));
+ Long earliest = nonnegativeLong(span.get("minTimeStamp"));
+ // 位点已过期或未初始化时，原生接口可能用 -1 时间戳计算出近似当前纪元的延迟。
+ if (queueId != null && delay != null && earliest != null && earliest > 0
+ && delay <= System.currentTimeMillis() - earliest) {
+ CollectRound.this.setData(MAPPER.consumerLag(this.topic, this.group, queueId.toString(), delay));
+ }
+ }
+ }
+ }
+
+ /** 步骤五：采集 Topic 与消费组对应的队列位点和积压量。 */
+ private class ConsumeStatsCallback extends CollectCallback {
+ private final String topic;
+ private final String group;
+
+ private ConsumeStatsCallback(RemotingCommand request, String topic, String group) {
+ super(request);
+ this.topic = topic;
+ this.group = group;
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ ConsumeStats stats = RemotingSerializable.decode(
+ responseFuture.getResponseCommand().getBody(), ConsumeStats.class);
+ stats.getOffsetTable().forEach((queue, offset) -> {
+ if (!topic.equals(queue.getTopic())) {
+ return;
+ }
+ String queueId = String.valueOf(queue.getQueueId());
+
+ CollectRound.this.setData(MAPPER.consumerOffset(topic, group, queueId, offset));
+ });
+ if (!stats.getOffsetTable().isEmpty()) {
+ QueryConsumeTimeSpanRequestHeader header = new QueryConsumeTimeSpanRequestHeader();
+ header.setTopic(this.topic);
+ header.setGroup(this.group);
+ new LagCallback(RemotingCommand.createRequestCommand(RequestCode.QUERY_CONSUME_TIME_SPAN, header),
+ this.topic, this.group).execute();
+ }
+ }
+ }
+ }
+
+ /** 采集消费组总数、Broker 消息统计、存储积压、消息保留时间、磁盘使用率及剩余空间、线程池队列长度。 */
+ private class BrokerCollect {
+ private final BrokerConfigCallback config = new BrokerConfigCallback();
+ private final BrokerRuntimeCallback runtime = new BrokerRuntimeCallback();
+ private final GroupCountCallback groups = new GroupCountCallback();
+
+ private void start() {
+ this.config.execute();
+ this.runtime.execute();
+ this.groups.execute();
+ }
+
+ /** 采集 Broker 配置的消费组总数。 */
+ private class GroupCountCallback extends CollectCallback {
+ private GroupCountCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG,
+ new GetAllSubscriptionGroupRequestHeader()));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> body = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (body != null && body.get("subscriptionGroupTable") instanceof Map<?, ?> groups) {
+ CollectRound.this.setData(MAPPER.groupNumber((long) groups.size()));
+ }
+ }
+ }
+
+ private void collectStats(String cluster, boolean incoming) {
+ ViewBrokerStatsDataRequestHeader header = new ViewBrokerStatsDataRequestHeader();
+ header.setStatsName(incoming ? "BROKER_PUT_NUMS" : "BROKER_GET_NUMS");
+ header.setStatsKey(cluster);
+ new BrokerStatsCallback(RemotingCommand.createRequestCommand(RequestCode.VIEW_BROKER_STATS_DATA, header), incoming).execute();
+ }
+
+ /** 步骤一：获取 Broker 所属集群名，启动消息写入和取出统计采集。 */
+ private class BrokerConfigCallback extends CollectCallback {
+ private BrokerConfigCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CONFIG, null));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) throws Exception {
+ Properties properties = new Properties();
+ properties.load(new StringReader(new String(responseFuture.getResponseCommand().getBody(), StandardCharsets.UTF_8)));
+ String cluster = properties.getProperty("brokerClusterName");
+ if (cluster == null || cluster.isBlank()) {
+ return;
+ }
+ collectStats(cluster, true);
+ collectStats(cluster, false);
+ }
+ }
+
+ /** 步骤二：采集分钟、小时、日窗口的消息写入和取出数量及 TPS。 */
+ private class BrokerStatsCallback extends WindowStatsCallback {
+ private final boolean incoming;
+
+ private BrokerStatsCallback(RemotingCommand request, boolean incoming) {
+ super(request);
+ this.incoming = incoming;
+ }
+
+ @Override
+ protected void recordWindow(String window, long count, float rate) {
+ CollectRound.this.setData(this.incoming
+ ? MAPPER.brokerMessagesIn(window, count, rate) : MAPPER.brokerMessagesOut(window, count, rate));
+ }
+ }
+
+ /** 步骤三：采集存储分发和刷盘积压、消息保留时间、磁盘使用率及剩余空间、线程池队列长度。 */
+ private class BrokerRuntimeCallback extends CollectCallback {
+ private BrokerRuntimeCallback() {
+ super(RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null));
+ }
+
+ @Override
+ protected void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> response = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ if (response == null || !(response.get("table") instanceof Map<?, ?> table)) {
+ return;
+ }
+ Long messagesIn = nonnegativeLong(table.get("msgPutTotalTodayNow"));
+ if (messagesIn != null) {
+ CollectRound.this.setData(MAPPER.messagesInTotal(messagesIn));
+ }
+ Long bytesIn = nonnegativeLong(table.get("putMessageSizeTotal"));
+ if (bytesIn != null) {
+ CollectRound.this.setData(MAPPER.throughputInTotal(bytesIn));
+ }
+ Long messagesOut = nonnegativeLong(table.get("msgGetTotalTodayNow"));
+ if (messagesOut != null) {
+ CollectRound.this.setData(MAPPER.messagesOutTotal(messagesOut));
+ }
+ Long dispatch = nonnegativeLong(table.get("dispatchBehindBytes"));
+ if (dispatch != null) {
+ CollectRound.this.setData(MAPPER.dispatchBytes(dispatch));
+ }
+ this.collectFlush(table);
+ this.collectDisk(table);
+ Long earliest = nonnegativeLong(table.get("earliestMessageTimeStamp"));
+ long now = System.currentTimeMillis();
+ if (earliest != null && earliest > 0 && earliest <= now) {
+ CollectRound.this.setData(MAPPER.reserveTime(now - earliest));
+ }
+ for (String pool : new String[] {"send", "pull", "litePull", "query", "ack"}) {
+ this.collectThreadPool(table, pool, pool + "ThreadPoolQueueSize");
+ }
+ this.collectThreadPool(table, "endTransaction", "EndTransactionQueueSize");
+ }
+
+ private void collectDisk(Map<?, ?> table) {
+ table.forEach((key, value) -> {
+ if (!(key instanceof String name)) {
+ return;
+ }
+ String type;
+ String path = "";
+ if (name.equals("commitLogDiskRatio")) {
+ type = "commitlog_min";
+ } else if (name.startsWith("commitLogDiskRatio_")) {
+ type = "commitlog";
+ path = name.substring("commitLogDiskRatio_".length());
+ } else if (name.equals("consumeQueueDiskRatio")) {
+ type = "consumequeue";
+ } else {
+ return;
+ }
+ Float ratio = nonnegativeFloat(value);
+ if (ratio != null && ratio <= 1) {
+ CollectRound.this.setData(MAPPER.diskUsage(type, path, ratio));
+ }
+ });
+ Object capacity = table.get("commitLogDirCapacity");
+ if (capacity instanceof String text) {
+ Matcher matcher = DISK_CAPACITY.matcher(text.trim());
+ if (matcher.matches()) {
+ Long total = byteCount(matcher.group(1));
+ Long free = byteCount(matcher.group(2));
+ if (total != null && total > 0 && free != null && free <= total) {
+ CollectRound.this.setData(MAPPER.diskFreeBytes(free));
+ }
+ }
+ }
+ }
+
+ private void collectFlush(Map<?, ?> table) {
+ Long flush = byteCount(table.get("remainHowManyDataToFlush"));
+ // 启用暂存池时，原生刷盘积压还包括待提交字节；字段缺席表示未启用该分支。
+ Long commit = table.containsKey("remainHowManyDataToCommit")
+ ? byteCount(table.get("remainHowManyDataToCommit")) : Long.valueOf(0L);
+ if (flush != null && commit != null && flush <= Long.MAX_VALUE - commit) {
+ CollectRound.this.setData(MAPPER.flushBytes(flush + commit));
+ }
+ }
+
+ private void collectThreadPool(Map<?, ?> table, String pool, String key) {
+ Long size = nonnegativeLong(table.get(key));
+ if (size != null) {
+ CollectRound.this.setData(MAPPER.threadPool(pool, size));
+ }
+ }
+ }
+
+ private Long byteCount(Object value) {
+ if (!(value instanceof String text)) {
+ return null;
+ }
+ Matcher matcher = BYTE_COUNT.matcher(text.trim());
+ if (!matcher.matches()) {
+ return null;
+ }
+ String unit = matcher.group(2);
+ int exponent = "B".equals(unit) ? 0 : "KMGTPE".indexOf(unit.charAt(0)) + 1;
+ try {
+ // Broker 已将二进制字节数格式化并舍入，此处只能还原近似值，不能声称字节级精确。
+ return new BigDecimal(matcher.group(1)).multiply(BigDecimal.valueOf(1024).pow(exponent))
+ .setScale(0, RoundingMode.HALF_UP).longValueExact();
+ } catch (ArithmeticException e) {
+ return null;
+ }
+ }
+ }
+
+ private Long nonnegativeLong(Object value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ long number = Long.parseLong(value.toString());
+ return number >= 0 ? number : null;
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ private Float nonnegativeFloatOrZero(Object value) {
+ Float result = this.nonnegativeFloat(value);
+ return result == null ? 0f : result;
+ }
+
+ private Float nonnegativeFloat(Object value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ double number = Double.parseDouble(value.toString());
+ float result = (float) number;
+ return number >= 0 && Float.isFinite(result) ? result : null;
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+ /** 解析 Broker、Topic 和消费组的分钟、小时、日消息统计窗口。 */
+ private abstract class WindowStatsCallback extends CollectCallback {
+ private WindowStatsCallback(RemotingCommand request) {
+ super(request);
+ }
+
+ @Override
+ protected final void handleResponse(ResponseFuture responseFuture) {
+ Map<?, ?> stats = RemotingSerializable.decode(responseFuture.getResponseCommand().getBody(), Map.class);
+ this.collectWindow(stats, "statsMinute", "minute");
+ this.collectWindow(stats, "statsHour", "hour");
+ this.collectWindow(stats, "statsDay", "day");
+ }
+
+ private void collectWindow(Map<?, ?> stats, String key, String window) {
+ if (stats == null || !(stats.get(key) instanceof Map<?, ?> sample)
+ || !(sample.get("sum") instanceof Number count) || !(sample.get("tps") instanceof Number rate)) {
+ return;
+ }
+ try {
+ long value = Long.parseLong(count.toString());
+ float tps = rate.floatValue();
+ if (value >= 0 && Float.isFinite(tps) && rate.doubleValue() >= 0) {
+ this.recordWindow(window, value, tps);
+ }
+ } catch (NumberFormatException e) {
+ // 小数、溢出或格式无效的消息数不参与采集。
+ }
+ }
+
+ protected abstract void recordWindow(String window, long count, float rate);
+ }
+
+ /** 统一处理请求发送、响应校验、异常和完成计数，具体响应由各采集回调解析。 */
+ private abstract class CollectCallback implements InvokeCallback {
+ private final RemotingCommand request;
+ private final AtomicBoolean finished = new AtomicBoolean();
+
+ private CollectCallback(RemotingCommand request) {
+ this.request = request;
+ // 请求发送前 +1；子请求先计数，父请求处理结束后才 -1。
+ CollectRound.this.pending.incrementAndGet();
+ }
+
+ // 发送异常也进入失败回调，每个请求只扣减一次。
+ protected final void execute() {
+ executeSafely(this,
+ () -> client.invokeAsync(this.request, this.timeoutMillis(), this));
+ }
+
+ private long timeoutMillis() throws TimeoutException {
+ long remaining = CollectRound.this.deadline - System.currentTimeMillis();
+ if (CollectRound.this.closed || remaining <= 0) {
+ throw new TimeoutException("RocketMQ collection deadline reached");
+ }
+ return Math.min(3000, remaining);
+ }
+
+ @Override
+ public final void operationComplete(ResponseFuture responseFuture) {
+ if (!this.finished.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ if (CollectRound.this.closed) {
+ return;
+ }
+ RemotingCommand response = responseFuture.getResponseCommand();
+ if (response == null) {
+ throw new IllegalStateException("Missing RocketMQ response");
+ }
+ if (response.getCode() != ResponseCode.SUCCESS || response.getBody() == null || response.getBody().length == 0) {
+ return;
+ }
+ this.handleResponse(responseFuture);
+ } catch (Exception e) {
+ this.restoreInterrupt(e);
+ } finally {
+ this.completeRequest();
+ }
+ }
+
+ @Override
+ public final void operationFail(Throwable cause) {
+ if (!this.finished.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ this.restoreInterrupt(cause);
+ } finally {
+ this.completeRequest();
+ }
+ }
+
+ private void restoreInterrupt(Throwable cause) {
+ if (cause instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private void completeRequest() {
+ // 成功、失败都 -1；finished 防止重复回调重复扣减。
+ if (CollectRound.this.pending.decrementAndGet() == 0) {
+ CollectRound.this.completed.countDown();
+ }
+ }
+
+ protected abstract void handleResponse(ResponseFuture responseFuture) throws Exception;
+ }
+
}
+
+ private static void executeSafely(InvokeCallback callback, RequestAction action) {
+ try {
+ action.run();
+ } catch (Exception e) {
+ callback.operationFail(e);
+ }
+ }
+
+ @FunctionalInterface
+ private interface RequestAction {
+ void run() throws Exception;
+ }
+
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/iotdb/IotDBReportMetaHandler.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/iotdb/IotDBReportMetaHandler.java
index 358d1bbf..bd8c3441 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/iotdb/IotDBReportMetaHandler.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/iotdb/IotDBReportMetaHandler.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.dashboard.console.function.report.iotdb;
import org.apache.eventmesh.dashboard.console.function.report.annotation.AbstractReportMetaHandler;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
import java.lang.reflect.Field;
import java.time.LocalDateTime;
@@ -155,7 +156,7 @@ public class IotDBReportMetaHandler extends AbstractReportMetaHandler {
if (Objects.equals(filedName, "time")) {
stringBuilder.append(" time");
- } else if (filedName.endsWith("Id")) {
+ } else if (filedName.endsWith("Id") || field.isAnnotationPresent(ReportTag.class)) {
stringBuilder.append(" tag");
} else if (filedName.startsWith("value") || field.getDeclaringClass() == this.reportMeta.getClazz()) {
stringBuilder.append(" field");
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/Rocketmq2ProducerOffset.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/Rocketmq2ProducerOffset.java
index d89fcb60..893873ad 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/Rocketmq2ProducerOffset.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/Rocketmq2ProducerOffset.java
@@ -20,6 +20,7 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
import lombok.Data;
@@ -28,8 +29,21 @@ import lombok.EqualsAndHashCode;
@Data
@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_producer_offset",
- defaultViewType = ReportViewType.HISTOGRAM, tableName = "rocketmq_producer_offset",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_producer_offset",
comment = "topic offset")
public class Rocketmq2ProducerOffset extends RuntimeLongValue {
+ @ReportTag
+ private String topicName;
+
+ private String queueId;
+
+ private Long valueMinOffset;
+
+ private Long valueLastUpdateTime;
+
+ private Long valueMinOffsetSum;
+
+ private Long valueMaxOffsetSum;
+
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqBrokerMessagesIn.java
similarity index 69%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqBrokerMessagesIn.java
index 43d32a80..d5ab65f0 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqBrokerMessagesIn.java
@@ -20,20 +20,24 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
+/** Broker 原生滚动采样窗口的写入消息数及消息 TPS；继承的 value 为消息数/秒，并非累计总量。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_broker_messages_in",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_broker_messages_in",
+ comment = "Broker 滚动采样窗口写入消息数及消息 TPS（消息数/秒）")
+public class RocketmqBrokerMessagesIn extends RuntimeFloatValue {
- private String messageType;
+ @ReportTag
+ private String window;
+ /** 滚动采样窗口内的写入消息数，不是进程累计计数。 */
+ private Long valueWindowCount;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqBrokerMessagesOut.java
similarity index 67%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqBrokerMessagesOut.java
index 43d32a80..1b9abe6a 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqBrokerMessagesOut.java
@@ -20,20 +20,24 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
+/** Broker 原生滚动采样窗口的取出消息数及消息 TPS；继承的 value 为消息数/秒，不代表消费确认。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_broker_messages_out",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_broker_messages_out",
+ comment = "Broker 滚动采样窗口取出消息数及消息 TPS（消息数/秒），非消费确认数")
+public class RocketmqBrokerMessagesOut extends RuntimeFloatValue {
- private String messageType;
+ @ReportTag
+ private String window;
+ /** 滚动采样窗口内的取出消息数，不是累计计数或已确认消费数。 */
+ private Long valueWindowCount;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerConnectionNumber.java
similarity index 78%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerConnectionNumber.java
index 43d32a80..f85bb9e1 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerConnectionNumber.java
@@ -20,20 +20,22 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_connection_number",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_connection_number",
+ comment = "消费组在当前 Broker 上的连接数")
+public class RocketmqConsumerConnectionNumber extends RuntimeId {
- private String messageType;
+ @ReportTag
+ private String groupName;
+ private Long valueConnectionCount;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerFailedTps.java
similarity index 72%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerFailedTps.java
index 43d32a80..0641f00f 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerFailedTps.java
@@ -20,20 +20,27 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
+/** 客户端最近一分钟消费失败 TPS，消息数每秒。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_failed_tps",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_failed_tps",
+ comment = "客户端最近一分钟消费失败 TPS，消息数每秒")
+public class RocketmqConsumerFailedTps extends RuntimeFloatValue {
+
+ @ReportTag
+ private String topicName;
- private String messageType;
+ @ReportTag
+ private String groupName;
+ @ReportTag
+ private String clientId;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java
index 7f2a2f1f..14fa5022 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java
@@ -29,8 +29,8 @@ import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_group_number",
defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_group_number",
- comment = "消费者组数量")
+ comment = "当前 Broker 配置的消费组总数，包含系统组和离线组")
public class RocketmqConsumerGroupNumber extends RuntimeLongValue {
-}
+}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerLagLatency.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerLagLatency.java
index c1ecd53e..8f6ca0f3 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerLagLatency.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerLagLatency.java
@@ -20,11 +20,27 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.SubscribeId.SubscribeLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/** 最早未消费消息的积压时间，毫秒；无积压为零。 */
+@Data
+@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_lag_latency",
defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_lag_latency",
- comment = "消费处理延迟时间")
-public class RocketmqConsumerLagLatency extends SubscribeLongValue {
+ comment = "最早未消费消息的积压时间，毫秒；无积压为零")
+public class RocketmqConsumerLagLatency extends RuntimeLongValue {
+
+ @ReportTag
+ private String topicName;
+
+ @ReportTag
+ private String groupName;
+
+ @ReportTag
+ private String queueId;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerOffset.java
similarity index 71%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerOffset.java
index 43d32a80..383c995a 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerOffset.java
@@ -20,20 +20,32 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_offset",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_offset",
+ comment = "消费组队列位点及位点差")
+public class RocketmqConsumerOffset extends RuntimeId {
+
+ @ReportTag
+ private String topicName;
+
+ @ReportTag
+ private String groupName;
+
+ private String queueId;
+
+ private Long valueConsumerOffset;
- private String messageType;
+ private Long valueBrokerOffset;
+ /** Broker 位点减去已提交位点，下限为零；表示位点差。 */
+ private Long valueOffsetLag;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerProcessTime.java
similarity index 69%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerProcessTime.java
index 43d32a80..92fc9fac 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerProcessTime.java
@@ -20,20 +20,27 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
+/** 客户端消费平均耗时，毫秒；沿用客户端分钟窗口及小时窗口回退语义。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_process_time",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_process_time",
+ comment = "客户端消费平均耗时，毫秒；沿用客户端分钟窗口及小时窗口回退语义")
+public class RocketmqConsumerProcessTime extends RuntimeFloatValue {
+
+ @ReportTag
+ private String topicName;
- private String messageType;
+ @ReportTag
+ private String groupName;
+ @ReportTag
+ private String clientId;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerSuccessTps.java
similarity index 72%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerSuccessTps.java
index 7f2a2f1f..f752a287 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerSuccessTps.java
@@ -20,17 +20,27 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
+/** 客户端最近一分钟消费成功 TPS，消息数每秒。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_group_number",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_group_number",
- comment = "消费者组数量")
-public class RocketmqConsumerGroupNumber extends RuntimeLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_success_tps",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_success_tps",
+ comment = "客户端最近一分钟消费成功 TPS，消息数每秒")
+public class RocketmqConsumerSuccessTps extends RuntimeFloatValue {
-}
+ @ReportTag
+ private String topicName;
+
+ @ReportTag
+ private String groupName;
+ @ReportTag
+ private String clientId;
+
+}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqGroupMessagesOut.java
similarity index 66%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqGroupMessagesOut.java
index 43d32a80..123e1880 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqGroupMessagesOut.java
@@ -20,20 +20,30 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
+/** Topic 与消费组滚动窗口取出消息数和 TPS，非消费确认数；继承的 value 单位为消息数/秒。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_group_messages_out",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_group_messages_out",
+ comment = "Topic 与消费组滚动窗口取出消息数和 TPS，非消费确认数")
+public class RocketmqGroupMessagesOut extends RuntimeFloatValue {
+
+ @ReportTag
+ private String topicName;
+
+ @ReportTag
+ private String groupName;
- private String messageType;
+ @ReportTag
+ private String window;
+ /** 滚动采样窗口内的消息数，不是进程累计计数。 */
+ private Long valueWindowCount;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
index 43d32a80..e29e25fb 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
@@ -20,7 +20,7 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
@@ -29,11 +29,7 @@ import lombok.EqualsAndHashCode;
@Data
@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
-
- private String messageType;
-
-
+ defaultViewType = ReportViewType.COUNTER, tableName = "rocketmq_messages_in_total",
+ comment = "Broker 启动以来接收的消息数")
+public class RocketmqMessagesInTotal extends RuntimeLongValue {
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesOutTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesOutTotal.java
index 8b9c0b2d..48b4e183 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesOutTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesOutTotal.java
@@ -20,7 +20,7 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.SubscribeId;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
@@ -29,13 +29,7 @@ import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_out_total",
defaultViewType = ReportViewType.COUNTER, tableName = "rocketmq_messages_out_total",
- comment = "已就绪消息量")
-public class RocketmqMessagesOutTotal extends SubscribeId {
-
- private Long topicNum;
-
- private Long minOffset;
-
- private Long maxOffset;
+ comment = "Broker 启动以来从存储读取的消息条数")
+public class RocketmqMessagesOutTotal extends RuntimeLongValue {
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageDiskFreeBytes.java
similarity index 80%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageDiskFreeBytes.java
index 7f2a2f1f..22348062 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqConsumerGroupNumber.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageDiskFreeBytes.java
@@ -25,12 +25,12 @@ import org.apache.eventmesh.dashboard.console.function.report.model.base.Runtime
import lombok.Data;
import lombok.EqualsAndHashCode;
+/** Broker 存储根目录所在文件系统剩余字节数，格式化容量还原的近似值。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_consumer_group_number",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_consumer_group_number",
- comment = "消费者组数量")
-public class RocketmqConsumerGroupNumber extends RuntimeLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_storage_disk_free_bytes",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_storage_disk_free_bytes",
+ comment = "Broker 存储根目录所在文件系统剩余字节数，格式化容量还原的近似值")
+public class RocketmqStorageDiskFreeBytes extends RuntimeLongValue {
}
-
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageDiskUsage.java
similarity index 72%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageDiskUsage.java
index 43d32a80..3c62e600 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageDiskUsage.java
@@ -20,20 +20,24 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
+/** 存储文件系统使用率，范围 0 到 1；commitlog 汇总为各路径最小值。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_storage_disk_usage",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_storage_disk_usage",
+ comment = "存储文件系统使用率，范围 0 到 1；commitlog 汇总为各路径最小值")
+public class RocketmqStorageDiskUsage extends RuntimeFloatValue {
- private String messageType;
+ @ReportTag
+ private String storeType;
+ @ReportTag
+ private String path;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageMessageReserveTime.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageMessageReserveTime.java
index ddd5547f..42a84d14 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageMessageReserveTime.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqStorageMessageReserveTime.java
@@ -20,11 +20,11 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_storage_message_reserve_time",
defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_storage_message_reserve_time",
- comment = "储存层消息保存时间")
-public class RocketmqStorageMessageReserveTime extends RuntimeId {
+ comment = "当前时间与最早保留消息存储时间之差（毫秒），非配置的保留时长")
+public class RocketmqStorageMessageReserveTime extends RuntimeLongValue {
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThreadPoolWartermark.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThreadPoolWartermark.java
index 0f0df097..d3c1c065 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThreadPoolWartermark.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThreadPoolWartermark.java
@@ -20,11 +20,20 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_thread_pool_wartermark",
defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_thread_pool_wartermark",
- comment = "线程池排队数")
+ comment = "线程池当前队列长度，非历史最大值")
public class RocketmqThreadPoolWartermark extends RuntimeLongValue {
+ @ReportTag
+ private String poolName;
+
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThroughputInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThroughputInTotal.java
index f2874956..38fafc23 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThroughputInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqThroughputInTotal.java
@@ -20,7 +20,7 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.SubscribeId.SubscribeLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
@@ -29,7 +29,7 @@ import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_throughput_in_total",
defaultViewType = ReportViewType.COUNTER, tableName = "rocketmq_throughput_in_total",
- comment = "消息生产流入服务端的吞吐量")
-public class RocketmqThroughputInTotal extends SubscribeLongValue {
+ comment = "Broker 启动以来接收的消息字节数")
+public class RocketmqThroughputInTotal extends RuntimeLongValue {
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqTopicMessagesIn.java
similarity index 68%
copy from eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
copy to eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqTopicMessagesIn.java
index 43d32a80..3dba3300 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqMessagesInTotal.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqTopicMessagesIn.java
@@ -20,20 +20,27 @@ package org.apache.eventmesh.dashboard.console.function.report.model.rocketmq;
import org.apache.eventmesh.dashboard.common.enums.ClusterType;
import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
-import org.apache.eventmesh.dashboard.console.function.report.model.base.not.TopicId.TopicLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
import lombok.Data;
import lombok.EqualsAndHashCode;
-
+/** Topic 滚动窗口写入消息数和 TPS，非生命周期累计量；继承的 value 单位为消息数/秒。 */
@Data
@EqualsAndHashCode(callSuper = true)
-@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_messages_in_total",
- defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_messages_in_total",
- comment = "消息生产数量")
-public class RocketmqMessagesInTotal extends TopicLongValue {
+@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_topic_messages_in",
+ defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_topic_messages_in",
+ comment = "Topic 滚动窗口写入消息数和 TPS，非生命周期累计量")
+public class RocketmqTopicMessagesIn extends RuntimeFloatValue {
+
+ @ReportTag
+ private String topicName;
- private String messageType;
+ @ReportTag
+ private String window;
+ /** 滚动采样窗口内的消息数，不是进程累计计数。 */
+ private Long valueWindowCount;
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqTopicNumber.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqTopicNumber.java
index 968b43c8..24950666 100644
--- a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqTopicNumber.java
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/function/report/model/rocketmq/RocketmqTopicNumber.java
@@ -24,7 +24,7 @@ import org.apache.eventmesh.dashboard.console.function.report.model.base.Runtime
@ReportMeta(clusterType = ClusterType.STORAGE_ROCKETMQ, reportName = "rocketmq_topic_number",
defaultViewType = ReportViewType.GAUGE, tableName = "rocketmq_topic_number",
- comment = "主题数量")
+ comment = "当前 Broker 配置的 Topic 总数，包含系统、重试和死信 Topic")
public class RocketmqTopicNumber extends RuntimeLongValue {
}
diff --git a/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/mapstruct/report/RocketMQCollectMapper.java b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/mapstruct/report/RocketMQCollectMapper.java
new file mode 100644
index 00000000..6daea830
--- /dev/null
+++ b/eventmesh-dashboard-console/src/main/java/org/apache/eventmesh/dashboard/console/mapstruct/report/RocketMQCollectMapper.java
@@ -0,0 +1,194 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.dashboard.console.mapstruct.report;
+
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerConnectionNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerFailedTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerGroupNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerLagLatency;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerOffset;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerProcessTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerSuccessTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqGroupMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesOutTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskFreeBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskUsage;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDispatchBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageFlushBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageMessageReserveTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThreadPoolWartermark;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThroughputInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicNumber;
+
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
+
+import org.mapstruct.BeanMapping;
+import org.mapstruct.Mapper;
+import org.mapstruct.Mapping;
+import org.mapstruct.factory.Mappers;
+
+/** 将 Broker 采样转换为语义对应的报表模型，由模型注解指定目标表。 */
+@Mapper
+public interface RocketMQCollectMapper {
+ RocketMQCollectMapper INSTANCE = Mappers.getMapper(RocketMQCollectMapper.class);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "groupName", source = "group")
+ @Mapping(target = "valueConnectionCount", source = "count")
+ RocketmqConsumerConnectionNumber connections(String group, long count);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "queueId", source = "queue")
+ @Mapping(target = "value", source = "offset.maxOffset")
+ @Mapping(target = "valueMinOffset", source = "offset.minOffset")
+ @Mapping(target = "valueLastUpdateTime", source = "offset.lastUpdateTimestamp")
+ Rocketmq2ProducerOffset topicOffset(String topic, String queue, TopicOffset offset);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "queueId", constant = "")
+ @Mapping(target = "valueMinOffsetSum", source = "min")
+ @Mapping(target = "valueMaxOffsetSum", source = "max")
+ @Mapping(target = "valueLastUpdateTime", source = "lastUpdate")
+ Rocketmq2ProducerOffset aggregateOffset(String topic, long min, long max, long lastUpdate);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "groupName", source = "group")
+ @Mapping(target = "queueId", source = "queue")
+ @Mapping(target = "valueConsumerOffset", source = "offset.consumerOffset")
+ @Mapping(target = "valueBrokerOffset", source = "offset.brokerOffset")
+ @Mapping(target = "valueOffsetLag", expression = "java(Math.max(0L, offset.getBrokerOffset() - offset.getConsumerOffset()))")
+ RocketmqConsumerOffset consumerOffset(String topic, String group, String queue, OffsetWrapper offset);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "window", source = "window")
+ @Mapping(target = "valueWindowCount", source = "count")
+ @Mapping(target = "value", source = "rate")
+ RocketmqBrokerMessagesIn brokerMessagesIn(String window, long count, float rate);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "window", source = "window")
+ @Mapping(target = "valueWindowCount", source = "count")
+ @Mapping(target = "value", source = "rate")
+ RocketmqBrokerMessagesOut brokerMessagesOut(String window, long count, float rate);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "bytes")
+ RocketmqStorageDispatchBehindBytes dispatchBytes(Long bytes);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "bytes")
+ RocketmqStorageFlushBehindBytes flushBytes(Long bytes);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "age")
+ RocketmqStorageMessageReserveTime reserveTime(Long age);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "poolName", source = "pool")
+ @Mapping(target = "value", source = "size")
+ RocketmqThreadPoolWartermark threadPool(String pool, long size);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "count")
+ RocketmqTopicNumber topicNumber(Long count);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "count")
+ RocketmqConsumerGroupNumber groupNumber(Long count);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "value")
+ RocketmqMessagesInTotal messagesInTotal(Long value);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "value")
+ RocketmqMessagesOutTotal messagesOutTotal(Long value);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "value")
+ RocketmqThroughputInTotal throughputInTotal(Long value);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "window", source = "window")
+ @Mapping(target = "valueWindowCount", source = "count")
+ @Mapping(target = "value", source = "rate")
+ RocketmqTopicMessagesIn topicMessages(String topic, String window, long count, float rate);
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topic")
+ @Mapping(target = "groupName", source = "group")
+ @Mapping(target = "window", source = "window")
+ @Mapping(target = "valueWindowCount", source = "count")
+ @Mapping(target = "value", source = "rate")
+ RocketmqGroupMessagesOut groupMessages(String topic, String group, String window, long count, float rate);
+
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topicName")
+ @Mapping(target = "groupName", source = "groupName")
+ @Mapping(target = "queueId", source = "queueId")
+ @Mapping(target = "value", source = "value")
+ RocketmqConsumerLagLatency consumerLag(String topicName, String groupName, String queueId, Long value);
+
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topicName")
+ @Mapping(target = "groupName", source = "groupName")
+ @Mapping(target = "clientId", source = "clientId")
+ @Mapping(target = "value", source = "value")
+ RocketmqConsumerSuccessTps consumerSuccess(String topicName, String groupName, String clientId, Float value);
+
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topicName")
+ @Mapping(target = "groupName", source = "groupName")
+ @Mapping(target = "clientId", source = "clientId")
+ @Mapping(target = "value", source = "value")
+ RocketmqConsumerFailedTps consumerFailed(String topicName, String groupName, String clientId, Float value);
+
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "topicName", source = "topicName")
+ @Mapping(target = "groupName", source = "groupName")
+ @Mapping(target = "clientId", source = "clientId")
+ @Mapping(target = "value", source = "value")
+ RocketmqConsumerProcessTime consumerProcessTime(String topicName, String groupName, String clientId, Float value);
+
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "storeType", source = "storeType")
+ @Mapping(target = "path", source = "path")
+ @Mapping(target = "value", source = "value")
+ RocketmqStorageDiskUsage diskUsage(String storeType, String path, Float value);
+
+
+ @BeanMapping(ignoreByDefault = true)
+ @Mapping(target = "value", source = "value")
+ RocketmqStorageDiskFreeBytes diskFreeBytes(Long value);
+
+}
diff --git a/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/DataSyncHandlerTest.java b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/DataSyncHandlerTest.java
new file mode 100644
index 00000000..3eaddd37
--- /dev/null
+++ b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/DataSyncHandlerTest.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.dashboard.console.function.report.collect;
+
+import org.apache.eventmesh.dashboard.console.function.report.ReportEngine;
+import org.apache.eventmesh.dashboard.console.function.report.ReportHandlerManage;
+import org.apache.eventmesh.dashboard.console.function.report.collect.padding.PaddingService;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerConnectionNumber;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class DataSyncHandlerTest {
+
+ @Test
+ public void initializationUsesConfiguredEngine() throws Exception {
+ ReportEngine engine = Mockito.mock(ReportEngine.class);
+ ReportHandlerManage reports = new ReportHandlerManage();
+ reports.setReportEngine(engine);
+ CollectManage manager = new CollectManage();
+ manager.setReportHandlerManage(reports);
+ manager.init();
+ var field = CollectManage.class.getDeclaredField("dataSyncHandler");
+ field.setAccessible(true);
+ DataSyncHandler handler = (DataSyncHandler) field.get(manager);
+ // 空结果也经过真实批次交付，验证初始化接线，而不是测试中手动注入引擎。
+ handler.getDataSyncHandlerWrapper(1).sync(data(0));
+ Mockito.verify(engine).batchInsertByClass(Map.of());
+ }
+
+ @Test
+ public void reverseCompletionDeliversFlatIterableRowsOnce() {
+ DataSyncHandler handler = new DataSyncHandler();
+ ReportEngine engine = Mockito.mock(ReportEngine.class);
+ PaddingService padding = Mockito.mock(PaddingService.class);
+ handler.setReportEngine(engine);
+ handler.setPaddingService(padding);
+ var wrapper = handler.getDataSyncHandlerWrapper(3);
+ RestoreData first = data(0, 10, 11);
+ RestoreData empty = data(1);
+ RestoreData last = data(2, 20);
+ List<Long> observed = new ArrayList<>();
+ Mockito.doAnswer(call -> {
+ Map<Class<?>, List<Object>> batch = call.getArgument(0);
+ List<Object> rows = batch.get(RocketmqConsumerConnectionNumber.class);
+ Assertions.assertEquals(3, rows.size());
+ Assertions.assertNotNull(rows.get(2));
+ for (Object row : rows) {
+ observed.add(((RocketmqConsumerConnectionNumber) row).getValueConnectionCount());
+ }
+ Assertions.assertFalse(first.getDataMap().isEmpty(), "Do not recycle before engine consumes the batch");
+ return null;
+ }).when(engine).batchInsertByClass(Mockito.anyMap());
+ wrapper.sync(last);
+ wrapper.sync(empty);
+ Mockito.verifyNoInteractions(engine);
+ wrapper.sync(first);
+ Assertions.assertEquals(List.of(20L, 10L, 11L), observed);
+ Mockito.verify(engine).batchInsertByClass(Mockito.anyMap());
+ Mockito.verify(padding, Mockito.times(3)).padding(Mockito.any());
+ Assertions.assertTrue(first.getDataMap().isEmpty());
+ Assertions.assertTrue(last.getDataMap().isEmpty());
+ Assertions.assertTrue(wrapper.restoreDataList.isEmpty());
+ }
+
+ @Test
+ public void concurrentCompletionAllocatesUniqueIndicesAndLosesNoRows() throws Exception {
+ DataSyncHandler handler = new DataSyncHandler();
+ ReportEngine engine = Mockito.mock(ReportEngine.class);
+ handler.setReportEngine(engine);
+ handler.setPaddingService(Mockito.mock(PaddingService.class));
+ var wrapper = handler.getDataSyncHandlerWrapper(200);
+ Set<Integer> indices = ConcurrentHashMap.newKeySet();
+ Set<Long> observed = ConcurrentHashMap.newKeySet();
+ Mockito.doAnswer(call -> {
+ Map<Class<?>, List<Object>> batch = call.getArgument(0);
+ List<Object> rows = batch.get(RocketmqConsumerConnectionNumber.class);
+ Assertions.assertEquals(200, rows.size());
+ for (Object row : rows) {
+ observed.add(((RocketmqConsumerConnectionNumber) row).getValueConnectionCount());
+ }
+ return null;
+ }).when(engine).batchInsertByClass(Mockito.anyMap());
+ var executor = Executors.newFixedThreadPool(8);
+ try {
+ List<Future<?>> jobs = new ArrayList<>();
+ for (int i = 0; i < 200; i++) {
+ final int value = i;
+ jobs.add(executor.submit(() -> {
+ int index = wrapper.getIndex();
+ Assertions.assertTrue(indices.add(index));
+ wrapper.sync(data(index, value));
+ }));
+ }
+ for (Future<?> job : jobs) {
+ job.get(5, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ Assertions.assertEquals(200, indices.size());
+ Assertions.assertEquals(200, observed.size());
+ Mockito.verify(engine).batchInsertByClass(Mockito.anyMap());
+ }
+
+ @Test
+ public void failedWriteStillReleasesContainers() {
+ DataSyncHandler handler = new DataSyncHandler();
+ ReportEngine engine = Mockito.mock(ReportEngine.class);
+ handler.setReportEngine(engine);
+ handler.setPaddingService(Mockito.mock(PaddingService.class));
+ Mockito.doThrow(new IllegalStateException("write failed")).when(engine).batchInsertByClass(Mockito.anyMap());
+ var wrapper = handler.getDataSyncHandlerWrapper(1);
+ RestoreData data = data(0, 1);
+ Assertions.assertThrows(IllegalStateException.class, () -> wrapper.sync(data));
+ Assertions.assertTrue(data.getDataMap().isEmpty());
+ Assertions.assertTrue(wrapper.restoreDataList.isEmpty());
+ }
+
+ private RestoreData data(int index, long... values) {
+ RestoreData data = new RestoreData();
+ data.setIndex(index);
+ data.setAbstractCollect(Mockito.mock(AbstractCollect.class));
+ for (long value : values) {
+ RocketmqConsumerConnectionNumber row = new RocketmqConsumerConnectionNumber();
+ row.setValueConnectionCount(value);
+ data.setData(row);
+ }
+ return data;
+ }
+}
diff --git a/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/RocketMQCollectManualTest.java b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/RocketMQCollectManualTest.java
new file mode 100644
index 00000000..cd6cb820
--- /dev/null
+++ b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/RocketMQCollectManualTest.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.dashboard.console.function.report.collect;
+
+import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.ClusterMetadata;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
+import org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper;
+import org.apache.eventmesh.dashboard.console.function.report.collect.exporter.RocketMQCollect;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.OrganizationId;
+import org.apache.eventmesh.dashboard.core.function.SDK.ConfigManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.AbstractSimpleCreateSDKConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.NetAddress;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import com.alibaba.fastjson.JSON;
+
+import lombok.extern.slf4j.Slf4j;
+
+
+@Slf4j
+public class RocketMQCollectManualTest {
+
+ /** 在 IDE 中运行此方法：调用一次 CollectManage.collect，读取真实 Broker 并打印全部采集结果。 */
+ @Test
+ @DisplayName("手动执行采集任务并打印真实 Broker 指标")
+ @SuppressWarnings("unchecked")
+ public void collectOnceFromRealBrokerAndPrint() throws Exception {
+ RuntimeMetadata runtime = new RuntimeMetadata();
+ runtime.setId(System.nanoTime());
+ runtime.setClusterId(1L);
+ runtime.setClusterType(ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE);
+ ClusterMetadata cluster = new ClusterMetadata();
+ cluster.setId(runtime.getClusterId());
+ cluster.setOrganizationId(7L);
+ cluster.setClusterType(runtime.getClusterType());
+ AtomicInteger printed = new AtomicInteger();
+ RocketMQCollect exporter = new RocketMQCollect() {
+ @Override
+ protected void setData(OrganizationId data) {
+ super.setData(data);
+ log.info("Broker 指标={} 数据={}", data.getClass().getSimpleName(), JSON.toJSONString(data));
+ printed.incrementAndGet();
+ }
+ };
+ exporter.setRuntimeMetadata(runtime);
+ exporter.setClusterMetadata(cluster);
+
+ AbstractSimpleCreateSDKConfig config = ConfigManage.getInstance()
+ .getSimpleCreateSdkConfig(runtime.getClusterType(), SDKTypeEnum.ADMIN);
+ NetAddress address = new NetAddress();
+ address.setAddress(System.getProperty("rocketmq.collect.host", "127.0.0.1"));
+ address.setPort(Integer.getInteger("rocketmq.collect.port", 20911));
+ config.setNetAddress(address);
+ DefaultRemotingClient client = SDKManage.getInstance()
+ .createClient(SDKTypeEnum.ADMIN, runtime, config, runtime.getClusterType());
+
+ CollectManage collectManage = new CollectManage();
+ Map<String, Collect> collectors = (Map<String, Collect>) FieldUtils.readField(collectManage, "executeingCollectMap", true);
+ collectors.put(runtime.getUnique(), exporter);
+
+ // 测试接收器只通知采集完成，不调用 IoTDB 写入链路。
+ CountDownLatch completed = new CountDownLatch(1);
+ DataSyncHandler printHandler = Mockito.mock(DataSyncHandler.class);
+ DataSyncHandlerWrapper wrapper = Mockito.mock(DataSyncHandlerWrapper.class);
+ Mockito.when(printHandler.getDataSyncHandlerWrapper(1)).thenReturn(wrapper);
+ Mockito.doAnswer(call -> {
+ completed.countDown();
+ return null;
+ }).when(wrapper).sync(Mockito.any(RestoreData.class));
+ FieldUtils.writeField(collectManage, "dataSyncHandler", printHandler, true);
+
+ ExecutorService workers = (ExecutorService) FieldUtils.readField(collectManage, "threadPoolExecutor", true);
+ try {
+ RemotingCommand response = client.invokeSync(
+ RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null), 3000);
+ Assertions.assertNotNull(response, "Broker unavailable at " + address);
+ Assertions.assertEquals(ResponseCode.SUCCESS, response.getCode(), "Broker unavailable at " + address);
+
+ // 这一行就是 ReportHandlerManage 定时任务实际调用的采集方法。
+ collectManage.collect();
+ Assertions.assertTrue(completed.await(10, TimeUnit.SECONDS), "Collection did not finish");
+ Assertions.assertTrue(printed.get() > 0, "Broker collection returned no data");
+ } finally {
+ workers.shutdownNow();
+ client.shutdown();
+ SDKManage.getInstance().deleteClient(null, runtime.getUnique());
+ }
+ }
+}
diff --git a/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/RocketMQCollectWithClientManualTest.java b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/RocketMQCollectWithClientManualTest.java
new file mode 100644
index 00000000..529025c2
--- /dev/null
+++ b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/RocketMQCollectWithClientManualTest.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.dashboard.console.function.report.collect;
+
+import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.ClusterMetadata;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
+import org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper;
+import org.apache.eventmesh.dashboard.console.function.report.collect.exporter.RocketMQCollect;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.OrganizationId;
+import org.apache.eventmesh.dashboard.core.function.SDK.ConfigManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.AbstractSimpleCreateSDKConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.NetAddress;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import com.alibaba.fastjson.JSON;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Manual test that extends {@link RocketMQCollectManualTest} by adding producer and consumer activity
+ * before collection, so that consumer-side metrics (connection count, consume rate, offset lag, etc.)
+ * can be collected.
+ *
+ * <p>Prerequisites: start RocketMQ via {@code docker compose up -d} with broker listening on
+ * {@code 127.0.0.1:20911} and NameServer on {@code 127.0.0.1:9876}. No VM options required.
+ *
+ * <p>Note: the consumer may not consume messages when running against a Docker-based broker due to
+ * network addressing, but this does not affect collection — the consumer registers with the Broker,
+ * so connection count, offset lag, and other consumer-side metrics are still available for collection.
+ */
+@Slf4j
+public class RocketMQCollectWithClientManualTest {
+
+ static {
+ System.setProperty("rocketmq.remoting.vip.channel.enabled", "false");
+ }
+
+ private static final String HOST = "127.0.0.1";
+ private static final int BROKER_PORT = 20911;
+ private static final String NAMESERVER = HOST + ":9876";
+ private static final String TEST_TOPIC = "CollectClientTestTopic";
+ private static final String PRODUCER_GROUP = "CollectClientTestProducerGroup";
+ private static final String CONSUMER_GROUP = "CollectClientTestGroup";
+
+ @Test
+ @DisplayName("带生产者消费者的采集测试")
+ @SuppressWarnings("unchecked")
+ public void collectWithClientActivityAndPrint() throws Exception {
+
+ DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
+ producer.setNamesrvAddr(NAMESERVER);
+ producer.setSendMsgTimeout(10000);
+ producer.setRetryTimesWhenSendFailed(3);
+ producer.start();
+ log.info("生产者已启动, namesrv={}", NAMESERVER);
+
+ boolean initSent = false;
+ for (int retry = 0; retry < 10; retry++) {
+ try {
+ SendResult result = producer.send(new Message(TEST_TOPIC, "TagA",
+ "InitMessage".getBytes(StandardCharsets.UTF_8)));
+ log.info("初始化消息发送成功: msgId={}, status={}", result.getMsgId(), result.getSendStatus());
+ initSent = true;
+ break;
+ } catch (Exception e) {
+ log.warn("发送初始化消息失败(第{}次): {}", retry + 1, e.getMessage());
+ Thread.sleep(3000);
+ }
+ }
+ Assertions.assertTrue(initSent, "无法发送初始化消息, 请检查 RocketMQ 是否正常启动");
+ log.info("Topic={} 已创建", TEST_TOPIC);
+
+ Thread.sleep(3000);
+
+ AtomicInteger consumed = new AtomicInteger();
+ CountDownLatch consumeLatch = new CountDownLatch(10);
+
+ DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
+ consumer.setNamesrvAddr(NAMESERVER);
+ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+ consumer.subscribe(TEST_TOPIC, "*");
+ consumer.registerMessageListener((MessageListenerConcurrently) (List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
+ for (MessageExt msg : msgs) {
+ int count = consumed.incrementAndGet();
+ consumeLatch.countDown();
+ log.info("消费消息 #{}: msgId={}", count, msg.getMsgId());
+ }
+ return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+ });
+ consumer.start();
+ log.info("消费者已启动, group={}", CONSUMER_GROUP);
+ Thread.sleep(5000);
+ log.info("消费者已就绪");
+
+ log.info("开始发送 10 条消息...");
+ for (int i = 0; i < 10; i++) {
+ SendResult result = producer.send(new Message(TEST_TOPIC, "TagA",
+ ("TestMessage_" + i).getBytes(StandardCharsets.UTF_8)));
+ log.info("发送 #{}: msgId={}, status={}", i, result.getMsgId(), result.getSendStatus());
+ }
+ log.info("已发送 10 条消息到 Topic={}", TEST_TOPIC);
+
+ consumeLatch.await(15, TimeUnit.SECONDS);
+ log.info("消费结果: 已消费={} 条", consumed.get());
+
+ Thread.sleep(3000);
+
+ RuntimeMetadata runtime = new RuntimeMetadata();
+ runtime.setId(System.nanoTime());
+ runtime.setClusterId(1L);
+ runtime.setClusterType(ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE);
+ ClusterMetadata cluster = new ClusterMetadata();
+ cluster.setId(runtime.getClusterId());
+ cluster.setOrganizationId(7L);
+ cluster.setClusterType(runtime.getClusterType());
+
+ AtomicInteger printed = new AtomicInteger();
+ Map<String, AtomicInteger> metricCounts = new ConcurrentHashMap<>();
+
+ RocketMQCollect exporter = new RocketMQCollect() {
+ @Override
+ protected void setData(OrganizationId data) {
+ super.setData(data);
+ String name = data.getClass().getSimpleName();
+ metricCounts.computeIfAbsent(name, k -> new AtomicInteger()).incrementAndGet();
+ log.info("采集指标={} 数据={}", name, JSON.toJSONString(data));
+ printed.incrementAndGet();
+ }
+ };
+ exporter.setRuntimeMetadata(runtime);
+ exporter.setClusterMetadata(cluster);
+
+ AbstractSimpleCreateSDKConfig config = ConfigManage.getInstance()
+ .getSimpleCreateSdkConfig(runtime.getClusterType(), SDKTypeEnum.ADMIN);
+ NetAddress address = new NetAddress();
+ address.setAddress(HOST);
+ address.setPort(BROKER_PORT);
+ config.setNetAddress(address);
+ DefaultRemotingClient client = SDKManage.getInstance()
+ .createClient(SDKTypeEnum.ADMIN, runtime, config, runtime.getClusterType());
+
+ CollectManage collectManage = new CollectManage();
+ Map<String, Collect> collectors = (Map<String, Collect>) FieldUtils.readField(collectManage, "executeingCollectMap", true);
+ collectors.put(runtime.getUnique(), exporter);
+
+ CountDownLatch completed = new CountDownLatch(1);
+ DataSyncHandler printHandler = Mockito.mock(DataSyncHandler.class);
+ DataSyncHandlerWrapper wrapper = Mockito.mock(DataSyncHandlerWrapper.class);
+ Mockito.when(printHandler.getDataSyncHandlerWrapper(1)).thenReturn(wrapper);
+ Mockito.doAnswer(call -> {
+ completed.countDown();
+ return null;
+ }).when(wrapper).sync(Mockito.any(RestoreData.class));
+ FieldUtils.writeField(collectManage, "dataSyncHandler", printHandler, true);
+
+ ExecutorService workers = (ExecutorService) FieldUtils.readField(collectManage, "threadPoolExecutor", true);
+ try {
+ RemotingCommand response = client.invokeSync(
+ RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null), 3000);
+ Assertions.assertNotNull(response, "Broker unavailable at " + address);
+ Assertions.assertEquals(ResponseCode.SUCCESS, response.getCode(), "Broker unavailable at " + address);
+
+ collectManage.collect();
+ Assertions.assertTrue(completed.await(15, TimeUnit.SECONDS), "采集未在 15 秒内完成");
+ Assertions.assertTrue(printed.get() > 0, "采集结果为空");
+ log.info("采集完成, 共采集到 {} 条指标数据", printed.get());
+
+ log.info("====== 指标采集汇总 ======");
+ metricCounts.entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
+ .forEach(e -> log.info(" {} : {} 条", e.getKey(), e.getValue().get()));
+ log.info("==========================");
+
+ } finally {
+ workers.shutdownNow();
+ client.shutdown();
+ SDKManage.getInstance().deleteClient(null, runtime.getUnique());
+ consumer.shutdown();
+ producer.shutdown();
+ log.info("生产者、消费者已关闭");
+ }
+ }
+}
diff --git a/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollectTest.java b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollectTest.java
new file mode 100644
index 00000000..8338436f
--- /dev/null
+++ b/eventmesh-dashboard-console/src/test/java/org/apache/eventmesh/dashboard/console/function/report/collect/exporter/RocketMQCollectTest.java
@@ -0,0 +1,1891 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.dashboard.console.function.report.collect.exporter;
+
+import org.apache.eventmesh.dashboard.common.enums.ClusterType;
+import org.apache.eventmesh.dashboard.common.model.metadata.RuntimeMetadata;
+import org.apache.eventmesh.dashboard.console.function.report.ReportViewType;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMeta;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportMetaData;
+import org.apache.eventmesh.dashboard.console.function.report.annotation.ReportTag;
+import org.apache.eventmesh.dashboard.console.function.report.iotdb.IotDBReportMetaHandler;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeFloatValue;
+import org.apache.eventmesh.dashboard.console.function.report.model.base.RuntimeId.RuntimeLongValue;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqBrokerMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerConnectionNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerFailedTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerGroupNumber;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerLagLatency;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerOffset;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerProcessTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerSuccessTps;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqGroupMessagesOut;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqMessagesOutTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskFreeBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDiskUsage;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageDispatchBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageFlushBehindBytes;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqStorageMessageReserveTime;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThroughputInTotal;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqThreadPoolWartermark;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicMessagesIn;
+import org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqTopicNumber;
+import org.apache.eventmesh.dashboard.console.mapstruct.report.RocketMQCollectMapper;
+import org.apache.eventmesh.dashboard.core.function.SDK.ConfigManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKManage;
+import org.apache.eventmesh.dashboard.core.function.SDK.SDKTypeEnum;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.AbstractSimpleCreateSDKConfig;
+import org.apache.eventmesh.dashboard.core.function.SDK.config.NetAddress;
+import org.apache.eventmesh.dashboard.core.function.SDK.operation.rocketmq.RocketMQRemotingSDKOperation.DefaultRemotingClient;
+
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.header.CreateTopicRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.DeleteSubscriptionGroupRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.DeleteTopicRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ViewBrokerStatsDataRequestHeader;
+import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ProducerData;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.slf4j.LoggerFactory;
+
+
+public class RocketMQCollectTest {
+
+ /** 验证实际缓存容器接收并发样本，保留各模型的全部记录。 */
+ @Test
+ @DisplayName("线程安全缓存完整保存并发样本")
+ public void concurrentCacheWritesKeepAllSamples() throws Exception {
+ var cache = new org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData();
+ int threadCount = 8;
+ int samplesPerThread = 5000;
+ var executor = java.util.concurrent.Executors.newFixedThreadPool(threadCount);
+ var start = new java.util.concurrent.CountDownLatch(1);
+ try {
+ List<java.util.concurrent.Future<?>> jobs = new java.util.ArrayList<>();
+ for (int thread = 0; thread < threadCount; thread++) {
+ final int base = thread * samplesPerThread;
+ jobs.add(executor.submit(() -> {
+ try {
+ Assertions.assertTrue(start.await(2, java.util.concurrent.TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ for (int sample = 0; sample < samplesPerThread; sample++) {
+ long value = base + sample;
+ cache.setData(value % 2 == 0 ? RocketMQCollectMapper.INSTANCE.topicNumber(value)
+ : RocketMQCollectMapper.INSTANCE.groupNumber(value));
+ }
+ }));
+ }
+ start.countDown();
+ for (var job : jobs) {
+ job.get(5, java.util.concurrent.TimeUnit.SECONDS);
+ }
+ } finally {
+ start.countDown();
+ executor.shutdownNow();
+ }
+ int expected = threadCount * samplesPerThread;
+ Assertions.assertEquals(Set.of(RocketmqTopicNumber.class, RocketmqConsumerGroupNumber.class), cache.getDataMap().keySet());
+ Set<Long> values = new java.util.HashSet<>();
+ cache.getDataMap().forEach((model, samples) -> {
+ Assertions.assertEquals(expected / 2, samples.size(), model.getSimpleName());
+ for (Object sample : samples) {
+ RuntimeLongValue row = Assertions.assertInstanceOf(RuntimeLongValue.class, sample);
+ Assertions.assertEquals(model == RocketmqTopicNumber.class ? 0 : 1, row.getValue() % 2);
+ Assertions.assertTrue(values.add(row.getValue()), "Sample values must remain unique");
+ }
+ });
+ Assertions.assertEquals(expected, values.size());
+ LoggerFactory.getLogger(RocketMQCollectTest.class).info("实际缓存并发验证通过：线程数={}，样本数={}，全部样本完整且唯一",
+ threadCount, values.size());
+ }
+
+ /** 验证消费位点、积压量和连接数正确映射到独立模型，并保留大整数精度。 */
+ @Test
+ @DisplayName("消费位点与连接数映射")
+ public void mapsConsumerSamplesToDedicatedModels() {
+ var offset = new org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper();
+ offset.setBrokerOffset(9007199254740993L);
+ offset.setConsumerOffset(9007199254740991L);
+ RocketmqConsumerOffset row = RocketMQCollectMapper.INSTANCE.consumerOffset("orders", "buyers", "3", offset);
+ Assertions.assertEquals("orders", row.getTopicName());
+ Assertions.assertEquals("buyers", row.getGroupName());
+ Assertions.assertEquals("3", row.getQueueId());
+ Assertions.assertNull(org.apache.commons.lang3.reflect.FieldUtils.getField(RocketmqConsumerOffset.class, "topicId", true));
+ Assertions.assertNull(org.apache.commons.lang3.reflect.FieldUtils.getField(RocketmqConsumerOffset.class, "groupId", true));
+ Assertions.assertEquals(9007199254740993L, row.getValueBrokerOffset());
+ Assertions.assertEquals(9007199254740991L, row.getValueConsumerOffset());
+ Assertions.assertEquals(2L, row.getValueOffsetLag());
+ offset.setConsumerOffset(9007199254740994L);
+ Assertions.assertEquals(0L,
+ RocketMQCollectMapper.INSTANCE.consumerOffset("orders", "buyers", "3", offset).getValueOffsetLag());
+ RocketmqConsumerConnectionNumber connections = RocketMQCollectMapper.INSTANCE.connections("buyers", 2L);
+ Assertions.assertEquals("buyers", connections.getGroupName());
+ Assertions.assertEquals(2L, connections.getValueConnectionCount());
+ Assertions.assertEquals(0L, RocketMQCollectMapper.INSTANCE.connections("buyers", 0L).getValueConnectionCount());
+ }
+
+ /** 验证生产位点、消费位点和连接数使用各自的 GAUGE 表及字段。 */
+ @Test
+ @DisplayName("生产位点、消费位点和连接数表结构")
+ public void dedicatedModelsDescribeSeparateGaugeTables() {
+ assertReportTable(org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class,
+ "rocketmq_producer_offset", List.of("topic_name", "queue_id"),
+ List.of("value", "value_min_offset", "value_last_update_time", "value_min_offset_sum", "value_max_offset_sum"));
+ assertReportTable(RocketmqConsumerOffset.class, "rocketmq_consumer_offset",
+ List.of("topic_name", "group_name", "queue_id"),
+ List.of("value_consumer_offset", "value_broker_offset", "value_offset_lag"));
+ assertReportTable(RocketmqConsumerConnectionNumber.class, "rocketmq_consumer_connection_number",
+ List.of("group_name"), List.of("value_connection_count"));
+ }
+
+ /** 模拟 Broker 运行状态响应，验证三项启动以来累计值和实例 ID。 */
+ @Test
+ @DisplayName("Broker 启动以来累计值采集")
+ public void collectsBrokerLifetimeCountersFromRuntimeResponse() throws Exception {
+ Map<Class<?>, List<Object>> models = collectRuntime(Map.of(
+ "msgPutTotalTodayNow", "12345",
+ "putMessageSizeTotal", "67890",
+ "msgGetTotalTodayNow", "23456"));
+
+ RocketmqMessagesInTotal messagesIn = rows(models, RocketmqMessagesInTotal.class).get(0);
+ Assertions.assertEquals(12345L, messagesIn.getValue());
+ Assertions.assertNotNull(messagesIn.getRuntimeId());
+ RocketmqThroughputInTotal bytesIn = rows(models, RocketmqThroughputInTotal.class).get(0);
+ Assertions.assertEquals(67890L, bytesIn.getValue());
+ Assertions.assertEquals(messagesIn.getRuntimeId(), bytesIn.getRuntimeId());
+ RocketmqMessagesOutTotal messagesOut = rows(models, RocketmqMessagesOutTotal.class).get(0);
+ Assertions.assertEquals(23456L, messagesOut.getValue());
+ Assertions.assertEquals(messagesIn.getRuntimeId(), messagesOut.getRuntimeId());
+ }
+
+ /** 验证合法零值会保留，缺失或非法的累计值不会生成指标。 */
+ @Test
+ @DisplayName("累计值的零值、缺失值与非法值")
+ public void keepsZeroCountersAndSkipsMissingOrInvalidRuntimeValues() throws Exception {
+ Map<Class<?>, List<Object>> zeros = collectRuntime(Map.of(
+ "msgPutTotalTodayNow", "0",
+ "putMessageSizeTotal", "0",
+ "msgGetTotalTodayNow", "0"));
+ Assertions.assertEquals(0L, rows(zeros, RocketmqMessagesInTotal.class).get(0).getValue());
+ Assertions.assertEquals(0L, rows(zeros, RocketmqThroughputInTotal.class).get(0).getValue());
+ Assertions.assertEquals(0L, rows(zeros, RocketmqMessagesOutTotal.class).get(0).getValue());
+
+ Map<Class<?>, List<Object>> invalid = collectRuntime(Map.of(
+ "msgPutTotalTodayNow", "-1",
+ "putMessageSizeTotal", "NaN",
+ "msgGetTotalTodayNow", "not-a-number"));
+ Assertions.assertFalse(invalid.containsKey(RocketmqMessagesInTotal.class));
+ Assertions.assertFalse(invalid.containsKey(RocketmqThroughputInTotal.class));
+ Assertions.assertFalse(invalid.containsKey(RocketmqMessagesOutTotal.class));
+
+ Map<Class<?>, List<Object>> missing = collectRuntime(Map.of());
+ Assertions.assertFalse(missing.containsKey(RocketmqMessagesInTotal.class));
+ Assertions.assertFalse(missing.containsKey(RocketmqThroughputInTotal.class));
+ Assertions.assertFalse(missing.containsKey(RocketmqMessagesOutTotal.class));
+ }
+
+ /** 验证三项累计指标是按实例区分的 COUNTER，数值字段为 int64。 */
+ @Test
+ @DisplayName("累计指标的 COUNTER 表结构")
+ public void lifetimeCountersHaveRuntimeScopedCounterSchemas() {
+ for (Class<?> model : List.of(RocketmqMessagesInTotal.class, RocketmqThroughputInTotal.class,
+ RocketmqMessagesOutTotal.class)) {
+ ReportMeta annotation = model.getAnnotation(ReportMeta.class);
+ Assertions.assertNotNull(annotation);
+ Assertions.assertEquals(ReportViewType.COUNTER, annotation.defaultViewType());
+ var metadata = new ReportMetaData();
+ metadata.setClazz(model);
+ metadata.setTableName(annotation.tableName());
+ metadata.setComment(annotation.comment());
+ var handler = new IotDBReportMetaHandler();
+ handler.setReportMeta(metadata);
+ handler.setFieldList(org.apache.commons.lang3.reflect.FieldUtils.getAllFieldsList(model));
+ String ddl = handler.createTable();
+ Assertions.assertTrue(ddl.contains("runtime_id string tag"), ddl);
+ Assertions.assertTrue(ddl.contains("value int64 field"), ddl);
+ Assertions.assertFalse(ddl.contains("topic_name"), ddl);
+ Assertions.assertFalse(ddl.contains("group_name"), ddl);
+ }
+ }
+
+ private void assertReportTable(Class<?> model, String table, List<String> tags, List<String> measurements) {
+ assertReportTable(model, table, tags, measurements, List.of());
+ }
+
+ private void assertReportTable(Class<?> model, String table, List<String> tags, List<String> measurements, List<String> floats) {
+ ReportMeta annotation = model.getAnnotation(ReportMeta.class);
+ Assertions.assertNotNull(annotation);
+ Assertions.assertEquals(table, annotation.tableName());
+ Assertions.assertEquals(table, annotation.reportName());
+ Assertions.assertEquals(ReportViewType.GAUGE, annotation.defaultViewType());
+ var metadata = new ReportMetaData();
+ metadata.setClazz(model);
+ metadata.setTableName(annotation.tableName());
+ metadata.setComment(annotation.comment());
+ var handler = new IotDBReportMetaHandler();
+ handler.setReportMeta(metadata);
+ handler.setFieldList(org.apache.commons.lang3.reflect.FieldUtils.getAllFieldsList(model));
+ String ddl = handler.createTable();
+ Assertions.assertTrue(ddl.startsWith("create table if not exists " + table + " "));
+ tags.forEach(tag -> Assertions.assertTrue(ddl.contains(tag + " string tag"), ddl));
+ measurements.forEach(field -> Assertions.assertTrue(ddl.contains(field + " int64 field"), ddl));
+ floats.forEach(field -> Assertions.assertTrue(ddl.contains(field + " float field"), ddl));
+ Assertions.assertTrue(ddl.contains("runtime_id string tag"), ddl);
+ }
+
+ /** 模拟消费响应，验证消费位点和连接数以独立模型交给父采集器。 */
+ @Test
+ @DisplayName("消费指标按独立模型交付")
+ public void handsConsumerSamplesToParentUnderDedicatedClasses() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ AtomicInteger connectionRequests = new AtomicInteger();
+ Mockito.doAnswer(invocation -> {
+ RemotingCommand request = invocation.getArgument(0);
+ InvokeCallback callback = invocation.getArgument(2);
+ if (failBrokerRoot(request, callback)) {
+ return null;
+ }
+ byte[] body;
+ switch (request.getCode()) {
+ case RequestCode.GET_ALL_TOPIC_CONFIG:
+ body = "{\"topicConfigTable\":{\"orders\":{},\"payments\":{}}}".getBytes(StandardCharsets.UTF_8);
+ break;
+ case RequestCode.GET_TOPIC_STATS_INFO:
+ body = RemotingSerializable.encode(new org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable());
+ break;
+ case RequestCode.QUERY_TOPIC_CONSUME_BY_WHO:
+ body = "{\"groupList\":[\"buyers\"]}".getBytes(StandardCharsets.UTF_8);
+ break;
+ case RequestCode.GET_CONSUMER_CONNECTION_LIST:
+ connectionRequests.incrementAndGet();
+ body = "{\"connectionSet\":[]}".getBytes(StandardCharsets.UTF_8);
+ break;
+ case RequestCode.GET_CONSUME_STATS:
+ var header = (org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader) request.readCustomHeader();
+ var stats = new org.apache.rocketmq.remoting.protocol.admin.ConsumeStats();
+ var offset = new org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper();
+ offset.setBrokerOffset(3L);
+ offset.setConsumerOffset(2L);
+ stats.getOffsetTable().put(new org.apache.rocketmq.common.message.MessageQueue(header.getTopic(), "broker", 0), offset);
+ body = RemotingSerializable.encode(stats);
+ break;
+ default:
+ throw new AssertionError("Unexpected request " + request.getCode());
+ }
+ reply(callback, ResponseCode.SUCCESS, body);
+ return null;
+ }).when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ collector(runtime).collect(0, wrapper);
+ }
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper).sync(capture.capture());
+ var models = capture.getValue().getDataMap();
+ Assertions.assertEquals(Set.of(RocketmqConsumerOffset.class, RocketmqConsumerConnectionNumber.class,
+ org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class, RocketmqTopicNumber.class), models.keySet());
+ Assertions.assertEquals(2, models.get(RocketmqConsumerOffset.class).size());
+ models.get(RocketmqConsumerOffset.class).stream().map(RocketmqConsumerOffset.class::cast).forEach(row -> {
+ Assertions.assertEquals("buyers", row.getGroupName());
+ Assertions.assertEquals(2L, row.getValueConsumerOffset());
+ Assertions.assertEquals(3L, row.getValueBrokerOffset());
+ Assertions.assertEquals(1L, row.getValueOffsetLag());
+ Assertions.assertEquals(runtime.getId(), row.getRuntimeId());
+ Assertions.assertEquals(7L, row.getOrganizationId());
+ });
+ Assertions.assertEquals(1, connectionRequests.get());
+ Assertions.assertEquals(1, models.get(RocketmqConsumerConnectionNumber.class).size());
+ }
+
+ /** 验证 Topic 的嵌套异步请求结束后，父采集器才交付本轮结果。 */
+ @Test
+ @DisplayName("等待嵌套请求完成再交付")
+ public void waitsForNestedRequestsThenHandsModelsToParent() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ RocketMQCollect collector = collector(runtime);
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ var calls = new java.util.concurrent.LinkedBlockingQueue<Call>();
+ var started = new java.util.concurrent.CountDownLatch(1);
+ Mockito.doAnswer(invocation -> {
+ if (failBrokerRoot(invocation.getArgument(0), invocation.getArgument(2))) {
+ return null;
+ }
+ calls.add(new Call(invocation.getArgument(0), invocation.getArgument(2)));
+ started.countDown();
+ return null;
+ }).when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ var executor = java.util.concurrent.Executors.newSingleThreadExecutor();
+ try {
+ var finished = executor.submit(() -> {
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ collector.collect(0, wrapper);
+ }
+ });
+ Assertions.assertTrue(started.await(2, java.util.concurrent.TimeUnit.SECONDS));
+ Call root = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertNotNull(root);
+ Assertions.assertEquals(RequestCode.GET_ALL_TOPIC_CONFIG, root.request().getCode());
+ reply(root.callback(), ResponseCode.SUCCESS,
+ "{\"topicConfigTable\":{\"test-topic\":{\"topicName\":\"test-topic\"}}}".getBytes(StandardCharsets.UTF_8));
+ Call offset = null;
+ for (int i = 0; i < 2; i++) {
+ Call child = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertNotNull(child);
+ if (child.request().getCode() == RequestCode.GET_TOPIC_STATS_INFO) {
+ offset = child;
+ } else {
+ reply(child.callback(), ResponseCode.SYSTEM_ERROR, null);
+ child.callback().operationFail(new IllegalStateException("duplicate completion"));
+ }
+ }
+ Assertions.assertNotNull(offset);
+ Assertions.assertFalse(finished.isDone(), "Nested Topic request must still be awaited");
+ Mockito.verifyNoInteractions(wrapper);
+ var stats = new org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable();
+ var position = new org.apache.rocketmq.remoting.protocol.admin.TopicOffset();
+ position.setMinOffset(1L);
+ position.setMaxOffset(9007199254740993L);
+ position.setLastUpdateTimestamp(123L);
+ stats.getOffsetTable().put(new org.apache.rocketmq.common.message.MessageQueue("test-topic", "broker", 0), position);
+ reply(offset.callback(), ResponseCode.SUCCESS, RemotingSerializable.encode(stats));
+ finished.get(2, java.util.concurrent.TimeUnit.SECONDS);
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper).sync(capture.capture());
+ var rows = capture.getValue().getDataMap().get(
+ org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class);
+ var queue = rows.stream().map(row ->
+ (org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset) row)
+ .filter(row -> "0".equals(row.getQueueId())).findFirst().orElseThrow();
+ Assertions.assertEquals(9007199254740993L, queue.getValue());
+ Assertions.assertEquals(1L, queue.getValueMinOffset());
+ Assertions.assertEquals(123L, queue.getValueLastUpdateTime());
+ Assertions.assertEquals(runtime.getId(), queue.getRuntimeId());
+ Assertions.assertEquals(7L, queue.getOrganizationId());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /** 验证采集超时后到达的回调不会继续写入本轮结果。 */
+ @Test
+ @DisplayName("超时后的回调不写入结果")
+ public void timeoutDoesNotLetLateCallbacksWriteIntoParent() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ RocketMQCollect collector = collector(runtime);
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ List<InvokeCallback> callbacks = new CopyOnWriteArrayList<>();
+ Mockito.doAnswer(call -> {
+ if (failBrokerRoot(call.getArgument(0), call.getArgument(2))) {
+ return null;
+ }
+ callbacks.add(call.getArgument(2));
+ return null;
+ }).when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ Assertions.assertTimeout(java.time.Duration.ofSeconds(6), () -> collector.collect(0, wrapper));
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper).sync(capture.capture());
+ Assertions.assertTrue(capture.getValue().getDataMap().isEmpty());
+ reply(callbacks.get(0), ResponseCode.SUCCESS, "{\"topicConfigTable\":{}}".getBytes(StandardCharsets.UTF_8));
+ Assertions.assertTrue(capture.getValue().getDataMap().isEmpty());
+ Mockito.verify(wrapper, Mockito.times(1)).sync(Mockito.any());
+ }
+ }
+
+ /** 验证已进入解析的旧回调在关闭后不调用父类写入，也不发送子请求。 */
+ @Test
+ @DisplayName("解析中的回调在关闭后放弃写入和子请求")
+ public void processingCallbackChecksClosedBeforeWritingAndSending() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ AtomicInteger writes = new AtomicInteger();
+ RocketMQCollect collector = new RocketMQCollect() {
+ @Override
+ protected void setData(org.apache.eventmesh.dashboard.console.function.report.model.base.OrganizationId data) {
+ writes.incrementAndGet();
+ super.setData(data);
+ }
+ };
+ var cluster = new org.apache.eventmesh.dashboard.common.model.metadata.ClusterMetadata();
+ cluster.setId(runtime.getClusterId());
+ cluster.setOrganizationId(7L);
+ cluster.setClusterType(runtime.getClusterType());
+ collector.setRuntimeMetadata(runtime);
+ collector.setClusterMetadata(cluster);
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ var calls = new java.util.concurrent.LinkedBlockingQueue<Call>();
+ AtomicInteger sends = new AtomicInteger();
+ var rootsSent = new java.util.concurrent.CountDownLatch(8);
+ Mockito.doAnswer(invocation -> {
+ sends.incrementAndGet();
+ rootsSent.countDown();
+ if (!failBrokerRoot(invocation.getArgument(0), invocation.getArgument(2))) {
+ calls.add(new Call(invocation.getArgument(0), invocation.getArgument(2)));
+ }
+ return null;
+ }).when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ var parsing = new java.util.concurrent.CountDownLatch(1);
+ var resume = new java.util.concurrent.CountDownLatch(1);
+ var executor = java.util.concurrent.Executors.newFixedThreadPool(2);
+ Runnable collect = () -> {
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ collector.collect(0, wrapper);
+ }
+ };
+ try {
+ var first = executor.submit(collect);
+ Call root = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertNotNull(root);
+ byte[] body = "{\"topicConfigTable\":{\"orders\":{}}}".getBytes(StandardCharsets.UTF_8);
+ RemotingCommand response = Mockito.mock(RemotingCommand.class);
+ Mockito.when(response.getCode()).thenReturn(ResponseCode.SUCCESS);
+ AtomicInteger reads = new AtomicInteger();
+ Mockito.when(response.getBody()).thenAnswer(invocation -> {
+ // 前两次读取属于公共校验，第三次读取已经进入 Topic 响应解析。
+ if (reads.incrementAndGet() == 3) {
+ parsing.countDown();
+ Assertions.assertTrue(resume.await(8, java.util.concurrent.TimeUnit.SECONDS));
+ }
+ return body;
+ });
+ ResponseFuture future = Mockito.mock(ResponseFuture.class);
+ Mockito.when(future.getResponseCommand()).thenReturn(response);
+ var oldCallback = executor.submit(() -> root.callback().operationComplete(future));
+ Assertions.assertTrue(parsing.await(2, java.util.concurrent.TimeUnit.SECONDS));
+ first.get(6, java.util.concurrent.TimeUnit.SECONDS);
+ var second = executor.submit(collect);
+ Call nextRoot = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertNotNull(nextRoot);
+ Assertions.assertTrue(rootsSent.await(2, java.util.concurrent.TimeUnit.SECONDS));
+ int sentBeforeResume = sends.get();
+ resume.countDown();
+ oldCallback.get(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertEquals(0, writes.get(), "Closed round must not invoke parent setData");
+ Assertions.assertEquals(sentBeforeResume, sends.get(), "Closed round must not send child requests");
+ var current = (org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData)
+ org.apache.commons.lang3.reflect.FieldUtils.readField(collector, "current", true);
+ Assertions.assertTrue(current.getDataMap().isEmpty(), "Old samples must not enter the next round");
+ reply(nextRoot.callback(), ResponseCode.SUCCESS, "{\"topicConfigTable\":{}}".getBytes(StandardCharsets.UTF_8));
+ second.get(2, java.util.concurrent.TimeUnit.SECONDS);
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper, Mockito.times(2)).sync(capture.capture());
+ Assertions.assertTrue(capture.getAllValues().get(0).getDataMap().isEmpty());
+ Assertions.assertEquals(1, writes.get());
+ Assertions.assertEquals(0L, rows(capture.getAllValues().get(1).getDataMap(), RocketmqTopicNumber.class).get(0).getValue());
+ LoggerFactory.getLogger(RocketMQCollectTest.class).info("模拟回调验证通过：解析期间超时关闭，旧回调未调用父类写入或发送子请求，下一轮正常采集");
+ } finally {
+ resume.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ /** 验证当前轮回调直接交付数据，上一轮迟到回调不会污染下一轮。 */
+ @Test
+ @DisplayName("直接交付并隔离前一轮回调")
+ public void publishesDirectlyAndRejectsPreviousRoundCallbacks() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ RocketMQCollect collector = collector(runtime);
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ var calls = new java.util.concurrent.LinkedBlockingQueue<Call>();
+ Mockito.doAnswer(invocation -> {
+ if (failBrokerRoot(invocation.getArgument(0), invocation.getArgument(2))) {
+ return null;
+ }
+ calls.add(new Call(invocation.getArgument(0), invocation.getArgument(2)));
+ return null;
+ }).when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ var executor = java.util.concurrent.Executors.newSingleThreadExecutor();
+ Runnable collect = () -> {
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ collector.collect(0, wrapper);
+ }
+ };
+ try {
+ var first = executor.submit(collect);
+ Call root = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertNotNull(root);
+ reply(root.callback(), ResponseCode.SUCCESS, "{\"topicConfigTable\":{\"orders\":{}}}".getBytes(StandardCharsets.UTF_8));
+ Call stats = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Call groups = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertNotNull(stats);
+ Assertions.assertNotNull(groups);
+ Assertions.assertEquals(RequestCode.GET_TOPIC_STATS_INFO, stats.request().getCode());
+ reply(stats.callback(), ResponseCode.SUCCESS,
+ RemotingSerializable.encode(new org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable()));
+ var current = (org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData)
+ org.apache.commons.lang3.reflect.FieldUtils.readField(collector, "current", true);
+ Assertions.assertEquals(1, current.getDataMap().get(org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class).size(),
+ "Completed metrics must reach setData before the remaining callback finishes");
+ Assertions.assertFalse(first.isDone());
+ first.get(6, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertEquals(1, current.getDataMap().get(org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class).size(),
+ "Timeout retains samples already delivered through setData");
+ var second = executor.submit(collect);
+ Call nextRoot = calls.poll(2, java.util.concurrent.TimeUnit.SECONDS);
+ Assertions.assertNotNull(nextRoot);
+ reply(groups.callback(), ResponseCode.SUCCESS, "{\"groupList\":[\"late-group\"]}".getBytes(StandardCharsets.UTF_8));
+ Assertions.assertTrue(calls.isEmpty(), "Previous round must not dispatch requests into a new round");
+ reply(nextRoot.callback(), ResponseCode.SUCCESS, "{\"topicConfigTable\":{}}".getBytes(StandardCharsets.UTF_8));
+ second.get(2, java.util.concurrent.TimeUnit.SECONDS);
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper, Mockito.times(2)).sync(capture.capture());
+ Assertions.assertEquals(0L, rows(capture.getAllValues().get(1).getDataMap(), RocketmqTopicNumber.class).get(0).getValue());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /** Offset regressions deliberately keep their callback queues independent of the new Broker roots. */
+ private boolean failBrokerRoot(RemotingCommand request, InvokeCallback callback) {
+ if (request.getCode() == RequestCode.GET_BROKER_CONFIG || request.getCode() == RequestCode.GET_BROKER_RUNTIME_INFO
+ || request.getCode() == RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG
+ || request.getCode() == RequestCode.VIEW_BROKER_STATS_DATA
+ || request.getCode() == RequestCode.QUERY_CONSUME_TIME_SPAN) {
+ callback.operationFail(new IllegalStateException("Broker roots disabled in offset-only regression"));
+ return true;
+ }
+ return false;
+ }
+
+ private void reply(InvokeCallback callback, int code, byte[] body) {
+ RemotingCommand response = RemotingCommand.createResponseCommand(code, "test response");
+ response.setBody(body);
+ ResponseFuture future = Mockito.mock(ResponseFuture.class);
+ Mockito.when(future.getResponseCommand()).thenReturn(response);
+ callback.operationComplete(future);
+ }
+
+ private record Call(RemotingCommand request, InvokeCallback callback) {
+ }
+
+ private RocketMQCollect collector(RuntimeMetadata runtime) {
+ var cluster = new org.apache.eventmesh.dashboard.common.model.metadata.ClusterMetadata();
+ cluster.setId(runtime.getClusterId());
+ cluster.setOrganizationId(7L);
+ cluster.setClusterType(runtime.getClusterType());
+ RocketMQCollect collector = new RocketMQCollect();
+ collector.setRuntimeMetadata(runtime);
+ collector.setClusterMetadata(cluster);
+ return collector;
+ }
+
+ /** 验证发送请求失败时，采集轮次能结束而不等待回调。 */
+ @Test
+ @DisplayName("请求发送失败时结束采集")
+ public void sendFailureFinishesRoundWithoutWaitingForCallback() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ RocketMQCollect collector = collector(runtime);
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ Mockito.doThrow(new IllegalStateException("send failed"))
+ .when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ Assertions.assertTimeout(java.time.Duration.ofSeconds(1), () -> collector.collect(0, wrapper));
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper).sync(capture.capture());
+ Assertions.assertTrue(capture.getValue().getDataMap().isEmpty());
+ }
+ }
+
+ /** 验证发送过程被中断后，线程的中断标志得到保留。 */
+ @Test
+ @DisplayName("发送中断后保留中断标志")
+ public void interruptedSendRestoresInterruptFlag() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ RocketMQCollect collector = collector(runtime);
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ Mockito.doThrow(new InterruptedException("interrupted"))
+ .when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ collector.collect(0, wrapper);
+ Assertions.assertTrue(Thread.currentThread().isInterrupted());
+ Mockito.verify(wrapper).sync(Mockito.any());
+ } finally {
+ Thread.interrupted();
+ }
+ }
+
+ /** 验证 Broker 窗口速率、累计条数和运行状态数值转换不丢失大整数精度。 */
+ @Test
+ @DisplayName("Broker 速率和运行状态数值映射")
+ public void mapsBrokerRatesAndRuntimeValuesWithoutLosingLongPrecision() {
+ long exact = 9007199254740993L;
+ var incoming = RocketMQCollectMapper.INSTANCE.brokerMessagesIn("minute", exact, 1.25F);
+ Assertions.assertEquals("minute", incoming.getWindow());
+ Assertions.assertEquals(exact, incoming.getValueWindowCount());
+ Assertions.assertEquals(Float.valueOf(1.25F), incoming.getValue());
+ var outgoing = RocketMQCollectMapper.INSTANCE.brokerMessagesOut("day", Long.MAX_VALUE, 0F);
+ Assertions.assertEquals("day", outgoing.getWindow());
+ Assertions.assertEquals(Long.MAX_VALUE, outgoing.getValueWindowCount());
+ Assertions.assertEquals(Float.valueOf(0F), outgoing.getValue());
+ Assertions.assertEquals(exact, RocketMQCollectMapper.INSTANCE.dispatchBytes(exact).getValue());
+ Assertions.assertEquals(exact, RocketMQCollectMapper.INSTANCE.flushBytes(exact).getValue());
+ Assertions.assertEquals(exact, RocketMQCollectMapper.INSTANCE.reserveTime(exact).getValue());
+ var pool = RocketMQCollectMapper.INSTANCE.threadPool("endTransaction", exact);
+ Assertions.assertEquals("endTransaction", pool.getPoolName());
+ Assertions.assertEquals(exact, pool.getValue());
+ }
+
+ /** 验证 Broker 窗口、线程池标签以及整数和浮点数字段类型。 */
+ @Test
+ @DisplayName("Broker 窗口与线程池表结构")
+ public void brokerModelsDescribeWindowPoolAndNumericColumnTypes() throws Exception {
+ assertReportTable(RocketmqBrokerMessagesIn.class, "rocketmq_broker_messages_in", List.of("window"),
+ List.of("value_window_count"), List.of("value"));
+ assertReportTable(RocketmqBrokerMessagesOut.class, "rocketmq_broker_messages_out", List.of("window"),
+ List.of("value_window_count"), List.of("value"));
+ assertReportTable(RocketmqStorageDispatchBehindBytes.class, "rocketmq_storage_dispatch_behind_bytes",
+ List.of(), List.of("value"));
+ assertReportTable(RocketmqStorageFlushBehindBytes.class, "rocketmq_storage_flush_behind_bytes", List.of(), List.of("value"));
+ assertReportTable(RocketmqStorageMessageReserveTime.class, "rocketmq_storage_message_reserve_time", List.of(), List.of("value"));
+ assertReportTable(RocketmqThreadPoolWartermark.class, "rocketmq_thread_pool_wartermark", List.of("pool_name"), List.of("value"));
+ Assertions.assertNotNull(RocketmqBrokerMessagesIn.class.getDeclaredField("window").getAnnotation(ReportTag.class));
+ Assertions.assertNotNull(RocketmqBrokerMessagesOut.class.getDeclaredField("window").getAnnotation(ReportTag.class));
+ Assertions.assertNotNull(RocketmqThreadPoolWartermark.class.getDeclaredField("poolName").getAnnotation(ReportTag.class));
+ Assertions.assertTrue(RuntimeFloatValue.class.isAssignableFrom(RocketmqBrokerMessagesIn.class));
+ Assertions.assertTrue(RuntimeFloatValue.class.isAssignableFrom(RocketmqBrokerMessagesOut.class));
+ Assertions.assertTrue(RuntimeLongValue.class.isAssignableFrom(RocketmqStorageMessageReserveTime.class));
+ }
+
+ /** 验证根请求先注册，同步失败和后续 Broker 统计子请求都正确计入采集轮次。 */
+ @Test
+ @DisplayName("根请求注册与 Broker 子请求等待")
+ public void registersAllRootsBeforeSynchronousFailureAndWaitsForNestedBrokerStats() throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ var calls = new java.util.concurrent.LinkedBlockingQueue<Call>();
+ var sent = new java.util.concurrent.CountDownLatch(3);
+ AtomicInteger topicRoots = new AtomicInteger();
+ AtomicInteger configRoots = new AtomicInteger();
+ Mockito.doAnswer(invocation -> {
+ RemotingCommand request = invocation.getArgument(0);
+ InvokeCallback callback = invocation.getArgument(2);
+ long timeout = invocation.getArgument(1);
+ Assertions.assertTrue(timeout > 0 && timeout <= 3000);
+ if (request.getCode() == RequestCode.GET_ALL_TOPIC_CONFIG) {
+ topicRoots.incrementAndGet();
+ callback.operationFail(new IllegalStateException("first root fails inline"));
+ } else if (request.getCode() == RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG) {
+ callback.operationFail(new IllegalStateException("Group count unavailable"));
+ } else if (request.getCode() == RequestCode.GET_BROKER_CONFIG) {
+ configRoots.incrementAndGet();
+ reply(callback, ResponseCode.SUCCESS, "brokerClusterName=cluster-A\nbrokerName=wrong-key\n".getBytes(StandardCharsets.UTF_8));
+ } else {
+ calls.add(new Call(request, callback));
+ sent.countDown();
+ }
+ return null;
+ }).when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ var executor = java.util.concurrent.Executors.newSingleThreadExecutor();
+ try {
+ var finished = executor.submit(() -> {
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ collector(runtime).collect(0, wrapper);
+ }
+ });
+ Assertions.assertTrue(sent.await(2, java.util.concurrent.TimeUnit.SECONDS));
+ Assertions.assertEquals(1, topicRoots.get());
+ Assertions.assertEquals(1, configRoots.get());
+ Assertions.assertEquals(3, calls.size());
+ Call runtimeCall = calls.stream().filter(call -> call.request().getCode() == RequestCode.GET_BROKER_RUNTIME_INFO)
+ .findFirst().orElseThrow();
+ List<Call> stats = calls.stream().filter(call -> call.request().getCode() == RequestCode.VIEW_BROKER_STATS_DATA).toList();
+ Assertions.assertEquals(2, stats.size());
+ Assertions.assertThrows(java.util.concurrent.TimeoutException.class,
+ () -> finished.get(100, java.util.concurrent.TimeUnit.MILLISECONDS));
+ Mockito.verifyNoInteractions(wrapper);
+ reply(runtimeCall.callback(), ResponseCode.SUCCESS, runtimeBody(Map.of("dispatchBehindBytes", "17")));
+ reply(stats.get(0).callback(), ResponseCode.SUCCESS, "{\"statsMinute\":{\"sum\":0,\"tps\":0}}".getBytes(StandardCharsets.UTF_8));
+ stats.get(0).callback().operationFail(new IllegalStateException("duplicate stats completion"));
+ Assertions.assertThrows(java.util.concurrent.TimeoutException.class,
+ () -> finished.get(100, java.util.concurrent.TimeUnit.MILLISECONDS));
+ Mockito.verifyNoInteractions(wrapper);
+ reply(stats.get(1).callback(), ResponseCode.SUCCESS, "{\"statsMinute\":{\"sum\":2,\"tps\":0.5}}".getBytes(StandardCharsets.UTF_8));
+ finished.get(2, java.util.concurrent.TimeUnit.SECONDS);
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper).sync(capture.capture());
+ var models = capture.getValue().getDataMap();
+ Assertions.assertEquals(Set.of(RocketmqBrokerMessagesIn.class, RocketmqBrokerMessagesOut.class,
+ RocketmqStorageDispatchBehindBytes.class), models.keySet());
+ Assertions.assertEquals(17L, rows(models, RocketmqStorageDispatchBehindBytes.class).get(0).getValue());
+ Assertions.assertEquals(1, rows(models, RocketmqBrokerMessagesIn.class).size());
+ Assertions.assertEquals(1, rows(models, RocketmqBrokerMessagesOut.class).size());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /** 验证 Broker 统计使用原生集群名查询，保留零值窗口和精确累计条数。 */
+ @Test
+ @DisplayName("Broker 原生集群键与窗口统计")
+ public void brokerStatsUseNativeClusterKeyAndKeepZeroWindowsAndExactCounts() throws Exception {
+ Set<String> names = new java.util.HashSet<>();
+ var models = collectMock((request, callback) -> {
+ switch (request.getCode()) {
+ case RequestCode.GET_BROKER_CONFIG:
+ reply(callback, ResponseCode.SUCCESS,
+ "brokerClusterName=集群-A\nbrokerName=not-the-cluster\n".getBytes(StandardCharsets.UTF_8));
+ break;
+ case RequestCode.VIEW_BROKER_STATS_DATA:
+ var header = (ViewBrokerStatsDataRequestHeader) request.readCustomHeader();
+ Assertions.assertEquals("集群-A", header.getStatsKey());
+ Assertions.assertTrue(names.add(header.getStatsName()), "Only one request per Broker statistic");
+ String body = "BROKER_PUT_NUMS".equals(header.getStatsName())
+ ? "{\"statsMinute\":{\"sum\":9007199254740993,\"tps\":1.25},\"statsHour\":{\"sum\":0,\"tps\":0}}"
+ : "{\"statsDay\":{\"sum\":9223372036854775807,\"tps\":0.5}}";
+ reply(callback, ResponseCode.SUCCESS, body.getBytes(StandardCharsets.UTF_8));
+ break;
+ default:
+ callback.operationFail(new IllegalStateException("Other collection direction unavailable"));
+ }
+ });
+ Assertions.assertEquals(Set.of("BROKER_PUT_NUMS", "BROKER_GET_NUMS"), names);
+ Assertions.assertEquals(Set.of(RocketmqBrokerMessagesIn.class, RocketmqBrokerMessagesOut.class), models.keySet());
+ var incoming = rows(models, RocketmqBrokerMessagesIn.class);
+ Assertions.assertEquals(2, incoming.size());
+ var minute = incoming.stream().filter(row -> "minute".equals(row.getWindow())).findFirst().orElseThrow();
+ Assertions.assertEquals(9007199254740993L, minute.getValueWindowCount());
+ Assertions.assertEquals(Float.valueOf(1.25F), minute.getValue());
+ var hour = incoming.stream().filter(row -> "hour".equals(row.getWindow())).findFirst().orElseThrow();
+ Assertions.assertEquals(0L, hour.getValueWindowCount());
+ Assertions.assertEquals(Float.valueOf(0F), hour.getValue());
+ var outgoing = rows(models, RocketmqBrokerMessagesOut.class);
+ Assertions.assertEquals(1, outgoing.size());
+ Assertions.assertEquals("day", outgoing.get(0).getWindow());
+ Assertions.assertEquals(Long.MAX_VALUE, outgoing.get(0).getValueWindowCount());
+ Assertions.assertEquals(Float.valueOf(0.5F), outgoing.get(0).getValue());
+ }
+
+ /** 验证非法窗口不会抹掉有效窗口，缺失窗口不会被补成零值。 */
+ @Test
+ @DisplayName("非法 Broker 窗口隔离")
+ public void invalidBrokerWindowsDoNotEraseValidWindowsOrInventZeros() throws Exception {
+ List<String> invalid = List.of("null", "{}", "[]", "\"invalid\"", "{\"sum\":1}", "{\"tps\":1}",
+ "{\"sum\":-1,\"tps\":1}", "{\"sum\":1.5,\"tps\":1}", "{\"sum\":9223372036854775808,\"tps\":1}",
+ "{\"sum\":\"1\",\"tps\":1}", "{\"sum\":null,\"tps\":1}", "{\"sum\":true,\"tps\":1}",
+ "{\"sum\":1,\"tps\":-1}", "{\"sum\":1,\"tps\":3.5e38}", "{\"sum\":1,\"tps\":1e309}",
+ "{\"sum\":1,\"tps\":\"NaN\"}", "{\"sum\":1,\"tps\":\"Infinity\"}",
+ "{\"sum\":1,\"tps\":\"0\"}", "{\"sum\":1,\"tps\":null}");
+ for (String window : invalid) {
+ var models = collectStatsBody("{\"statsMinute\":" + window
+ + ",\"statsHour\":{\"sum\":5,\"tps\":2.5},\"statsDay\":{\"sum\":0,\"tps\":0}}");
+ Assertions.assertEquals(Set.of(RocketmqBrokerMessagesIn.class, RocketmqBrokerMessagesOut.class), models.keySet(), window);
+ for (Class<?> type : List.of(RocketmqBrokerMessagesIn.class, RocketmqBrokerMessagesOut.class)) {
+ List<Object> samples = models.get(type);
+ Assertions.assertEquals(2, samples.size(), window);
+ Map<String, Long> counts = new java.util.HashMap<>();
+ for (Object row : samples) {
+ String name = brokerWindow(row);
+ counts.put(name, brokerWindowCount(row));
+ Assertions.assertEquals(Float.valueOf("hour".equals(name) ? 2.5F : 0F), ((RuntimeFloatValue) row).getValue(), window);
+ }
+ Assertions.assertEquals(Map.of("hour", 5L, "day", 0L), counts, window);
+ }
+ }
+ Assertions.assertTrue(collectStatsBody("{}").isEmpty(), "Missing windows are not zero-valued observations");
+ }
+
+ /** 验证集群名或统计数据不可用时，不会用位点推算 Broker 消息速率。 */
+ @Test
+ @DisplayName("Broker 统计缺失时不推算速率")
+ public void unavailableBrokerStatsAndClusterNameNeverProduceOffsetDerivedRates() throws Exception {
+ for (String config : List.of("", "brokerName=not-a-cluster\n", "brokerClusterName= \n")) {
+ AtomicInteger statsRequests = new AtomicInteger();
+ var models = collectMock((request, callback) -> {
+ if (request.getCode() == RequestCode.GET_BROKER_CONFIG) {
+ reply(callback, ResponseCode.SUCCESS, config.getBytes(StandardCharsets.UTF_8));
+ } else if (request.getCode() == RequestCode.GET_BROKER_RUNTIME_INFO) {
+ reply(callback, ResponseCode.SUCCESS, runtimeBody(Map.of("dispatchBehindBytes", "3")));
+ } else if (request.getCode() == RequestCode.VIEW_BROKER_STATS_DATA) {
+ statsRequests.incrementAndGet();
+ callback.operationFail(new IllegalStateException("Must not guess a Broker stats key"));
+ } else {
+ callback.operationFail(new IllegalStateException("Topic unavailable"));
+ }
+ });
+ Assertions.assertEquals(0, statsRequests.get(), config);
+ Assertions.assertEquals(Set.of(RocketmqStorageDispatchBehindBytes.class), models.keySet(), config);
+ }
+ for (byte[] body : new byte[][] {null, "".getBytes(StandardCharsets.UTF_8), "bad-json".getBytes(StandardCharsets.UTF_8),
+ "[]".getBytes(StandardCharsets.UTF_8)}) {
+ Assertions.assertTrue(collectMock((request, callback) -> {
+ if (request.getCode() == RequestCode.GET_BROKER_CONFIG) {
+ reply(callback, ResponseCode.SUCCESS, "brokerClusterName=cluster-A\n".getBytes(StandardCharsets.UTF_8));
+ } else if (request.getCode() == RequestCode.VIEW_BROKER_STATS_DATA) {
+ reply(callback, ResponseCode.SUCCESS, body);
+ } else {
+ callback.operationFail(new IllegalStateException("Other root unavailable"));
+ }
+ }).isEmpty(), "Malformed Broker responses must not produce default windows");
+ }
+ var models = collectMock((request, callback) -> {
+ switch (request.getCode()) {
+ case RequestCode.GET_BROKER_CONFIG:
+ reply(callback, ResponseCode.SUCCESS, "brokerClusterName=cluster-A\n".getBytes(StandardCharsets.UTF_8));
+ break;
+ case RequestCode.VIEW_BROKER_STATS_DATA:
+ reply(callback, ResponseCode.SYSTEM_ERROR, "{\"statsMinute\":{\"sum\":99,\"tps\":99}}".getBytes(StandardCharsets.UTF_8));
+ break;
+ case RequestCode.GET_ALL_TOPIC_CONFIG:
+ reply(callback, ResponseCode.SUCCESS, "{\"topicConfigTable\":{\"orders\":{}}}".getBytes(StandardCharsets.UTF_8));
+ break;
+ case RequestCode.GET_TOPIC_STATS_INFO:
+ var stats = new org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable();
+ var offset = new org.apache.rocketmq.remoting.protocol.admin.TopicOffset();
+ offset.setMinOffset(100);
+ offset.setMaxOffset(999);
+ stats.getOffsetTable().put(new org.apache.rocketmq.common.message.MessageQueue("orders", "broker", 0), offset);
+ reply(callback, ResponseCode.SUCCESS, RemotingSerializable.encode(stats));
+ break;
+ default:
+ callback.operationFail(new IllegalStateException("No runtime or consumers"));
+ }
+ });
+ Assertions.assertEquals(Set.of(org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class,
+ RocketmqTopicNumber.class), models.keySet(), "Topic positions must never become Broker message counts or rates");
+ var offsets = rows(models, org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class);
+ Assertions.assertEquals(2, offsets.size());
+ Assertions.assertTrue(offsets.stream().anyMatch(row -> Long.valueOf(999L).equals(row.getValueMaxOffsetSum())));
+ }
+
+ /** 验证运行状态原生字段、精确字节数及各线程池维度。 */
+ @Test
+ @DisplayName("Broker 运行状态原生字段解析")
+ public void runtimeUsesNativeKeysAndPreservesExactBytesAndAllPoolDimensions() throws Exception {
+ long earliest = System.currentTimeMillis() - 60000;
+ long before = System.currentTimeMillis();
+ var values = new java.util.HashMap<String, String>();
+ values.put("dispatchBehindBytes", "9007199254740993");
+ values.put("remainHowManyDataToFlush", " 1.5 KiB ");
+ values.put("remainHowManyDataToCommit", "0.5KiB");
+ values.put("earliestMessageTimeStamp", Long.toString(earliest));
+ values.put("sendThreadPoolQueueSize", "0");
+ values.put("pullThreadPoolQueueSize", "1");
+ values.put("litePullThreadPoolQueueSize", "2");
+ values.put("queryThreadPoolQueueSize", "3");
+ values.put("ackThreadPoolQueueSize", "4");
+ values.put("EndTransactionQueueSize", "9007199254740993");
+ values.put("putTps", "123 456 789");
+ values.put("getTps", "123 456 789");
+ values.put("putMessageTimesTotal", "1000");
+ var models = collectRuntime(values);
+ long after = System.currentTimeMillis();
+ Assertions.assertEquals(Set.of(RocketmqStorageDispatchBehindBytes.class, RocketmqStorageFlushBehindBytes.class,
+ RocketmqStorageMessageReserveTime.class, RocketmqThreadPoolWartermark.class), models.keySet());
+ Assertions.assertEquals(9007199254740993L, rows(models, RocketmqStorageDispatchBehindBytes.class).get(0).getValue());
+ Assertions.assertEquals(2048L, rows(models, RocketmqStorageFlushBehindBytes.class).get(0).getValue());
+ long age = rows(models, RocketmqStorageMessageReserveTime.class).get(0).getValue();
+ Assertions.assertTrue(age >= before - earliest && age <= after - earliest, "Reserve time is age in milliseconds");
+ Map<String, Long> pools = new java.util.HashMap<>();
+ rows(models, RocketmqThreadPoolWartermark.class).forEach(row -> {
+ Assertions.assertNull(pools.put(row.getPoolName(), row.getValue()), "One sample per native pool");
+ Assertions.assertEquals(7L, row.getOrganizationId());
+ Assertions.assertEquals(1L, row.getClustersId());
+ Assertions.assertNotNull(row.getRuntimeId());
+ Assertions.assertNotNull(row.getTime());
+ });
+ Assertions.assertEquals(Map.of("send", 0L, "pull", 1L, "litePull", 2L, "query", 3L, "ack", 4L,
+ "endTransaction", 9007199254740993L), pools);
+ }
+
+ /** 验证待刷盘字节的单位换算和取整，并拒绝溢出或格式错误。 */
+ @Test
+ @DisplayName("待刷盘字节单位与异常值")
+ public void flushBytesParseNativeUnitsRoundAndRejectOverflowOrMalformedCommit() throws Exception {
+ Map<String, Long> accepted = Map.ofEntries(Map.entry("0 B", 0L), Map.entry("0.5 B", 1L),
+ Map.entry("9007199254740993 B", 9007199254740993L), Map.entry("9223372036854775807 B", Long.MAX_VALUE),
+ Map.entry(" 1.5 KiB ", 1536L), Map.entry("1.5MiB", 1572864L), Map.entry("1 GiB", 1073741824L),
+ Map.entry("1 TiB", 1099511627776L), Map.entry("1 PiB", 1125899906842624L));
+ for (var entry : accepted.entrySet()) {
+ var models = collectRuntime(Map.of("remainHowManyDataToFlush", entry.getKey()));
+ Assertions.assertEquals(Set.of(RocketmqStorageFlushBehindBytes.class), models.keySet(), entry.getKey());
+ Assertions.assertEquals(entry.getValue(), rows(models, RocketmqStorageFlushBehindBytes.class).get(0).getValue(), entry.getKey());
+ }
+ var summed = collectRuntime(Map.of("remainHowManyDataToFlush", "9007199254740993 B", "remainHowManyDataToCommit", "2 B"));
+ Assertions.assertEquals(9007199254740995L, rows(summed, RocketmqStorageFlushBehindBytes.class).get(0).getValue());
+ List<Map<String, String>> invalid = List.of(
+ Map.of("remainHowManyDataToFlush", "9223372036854775807 B", "remainHowManyDataToCommit", "1 B"),
+ Map.of("remainHowManyDataToFlush", "9223372036854775808 B"), Map.of("remainHowManyDataToFlush", "8192 PiB"),
+ Map.of("remainHowManyDataToFlush", "-1 B"), Map.of("remainHowManyDataToFlush", "NaN B"),
+ Map.of("remainHowManyDataToFlush", ""), Map.of("remainHowManyDataToFlush", "nonsense"),
+ Map.of("remainHowManyDataToFlush", "1 B", "remainHowManyDataToCommit", "bad"),
+ Map.of("remainHowManyDataToFlush", "1 B", "remainHowManyDataToCommit", ""),
+ Map.of("remainHowManyDataToFlush", "1 B", "remainHowManyDataToCommit", "-1 B"),
+ Map.of("remainHowManyDataToCommit", "1 B"));
+ for (var fields : invalid) {
+ var values = new java.util.HashMap<>(fields);
+ values.put("dispatchBehindBytes", "5");
+ var models = collectRuntime(values);
+ Assertions.assertEquals(Set.of(RocketmqStorageDispatchBehindBytes.class), models.keySet(), fields.toString());
+ Assertions.assertEquals(5L, rows(models, RocketmqStorageDispatchBehindBytes.class).get(0).getValue());
+ }
+ }
+
+ /** 验证单个运行状态字段错误不会影响其他指标，也不会采用替代字段伪造结果。 */
+ @Test
+ @DisplayName("运行状态错误字段隔离")
+ public void malformedRuntimeFieldsDoNotEraseOtherMeasurementsOrUseFallbacks() throws Exception {
+ for (String invalid : List.of("-1", "1.5", "NaN", "9223372036854775808", "")) {
+ var models = collectRuntime(Map.of("dispatchBehindBytes", invalid, "sendThreadPoolQueueSize", invalid,
+ "remainHowManyDataToFlush", "2 B", "queryThreadPoolQueueSize", "7",
+ "earliestMessageTimeStamp", Long.toString(System.currentTimeMillis() + 3600000),
+ "putTps", "1 2 3", "getTps", "4 5 6"));
+ Assertions.assertEquals(Set.of(RocketmqStorageFlushBehindBytes.class, RocketmqThreadPoolWartermark.class),
+ models.keySet(), invalid);
+ Assertions.assertEquals(2L, rows(models, RocketmqStorageFlushBehindBytes.class).get(0).getValue());
+ var pools = rows(models, RocketmqThreadPoolWartermark.class);
+ Assertions.assertEquals(1, pools.size());
+ Assertions.assertEquals("query", pools.get(0).getPoolName());
+ Assertions.assertEquals(7L, pools.get(0).getValue());
+ }
+ for (String earliest : List.of("0", "-1", "bad", "1.5", "9223372036854775808")) {
+ Assertions.assertTrue(collectRuntime(Map.of("earliestMessageTimeStamp", earliest)).isEmpty(), earliest);
+ }
+ Assertions.assertTrue(collectRuntime(Map.of("putTps", "1 2 3", "getTps", "4 5 6",
+ "dispatchBehind", "9", "flushBehindBytes", "9", "endTransactionThreadPoolQueueSize", "9")).isEmpty(),
+ "Unknown keys and runtime throughput cannot substitute for native Broker statistics");
+ }
+
+ /** 验证消费积压时间过滤非法时间戳，并保留合法的零积压。 */
+ @Test
+ @DisplayName("消费积压时间与零积压")
+ public void filtersInvalidLagTimestampsAndRetainsZeroBacklog() throws Exception {
+ long now = System.currentTimeMillis();
+ for (long delay : new long[] {0, 100, -1, now + 1}) {
+ var models = collectMock((request, callback) -> {
+ switch (request.getCode()) {
+ case RequestCode.GET_ALL_TOPIC_CONFIG -> reply(callback, ResponseCode.SUCCESS,
+ "{\"topicConfigTable\":{\"orders\":{}}}".getBytes(StandardCharsets.UTF_8));
+ case RequestCode.QUERY_TOPIC_CONSUME_BY_WHO -> reply(callback, ResponseCode.SUCCESS,
+ "{\"groupList\":[\"buyers\"]}".getBytes(StandardCharsets.UTF_8));
+ case RequestCode.GET_CONSUME_STATS -> {
+ var stats = new org.apache.rocketmq.remoting.protocol.admin.ConsumeStats();
+ var offset = new org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper();
+ offset.setBrokerOffset(3L);
+ offset.setConsumerOffset(2L);
+ stats.getOffsetTable().put(new org.apache.rocketmq.common.message.MessageQueue("orders", "broker", 0), offset);
+ reply(callback, ResponseCode.SUCCESS, RemotingSerializable.encode(stats));
+ }
+ case RequestCode.QUERY_CONSUME_TIME_SPAN -> {
+ byte[] body = RemotingSerializable.encode(Map.of("consumeTimeSpanSet", List.of(
+ Map.of("messageQueue", Map.of("topic", "orders", "queueId", 0), "delayTime", delay, "minTimeStamp", now - 1000),
+ Map.of("messageQueue", Map.of("topic", "orders", "queueId", 1), "delayTime", 0, "minTimeStamp", -1),
+ Map.of("messageQueue", Map.of("topic", "another", "queueId", 0), "delayTime", 0, "minTimeStamp", now - 1000))));
+ reply(callback, ResponseCode.SUCCESS, body);
+ reply(callback, ResponseCode.SUCCESS, body);
+ }
+ default -> callback.operationFail(new IllegalStateException("Other requests unavailable"));
+ }
+ });
+ if (delay >= 0 && delay <= 100) {
+ var rows = rows(models, RocketmqConsumerLagLatency.class);
+ Assertions.assertEquals(1, rows.size(), "Duplicate callbacks must not duplicate samples");
+ Assertions.assertEquals(delay, rows.get(0).getValue());
+ Assertions.assertEquals("buyers", rows.get(0).getGroupName());
+ Assertions.assertEquals("0", rows.get(0).getQueueId());
+ } else {
+ Assertions.assertFalse(models.containsKey(RocketmqConsumerLagLatency.class));
+ }
+ }
+ }
+
+ /** 验证磁盘使用率和近似剩余字节采集，缺失或非法值保持缺失。 */
+ @Test
+ @DisplayName("磁盘使用率与剩余空间采集")
+ public void collectsDiskRatiosAndApproximateFreeBytesWithoutInventingMissingValues() throws Exception {
+ var models = collectRuntime(Map.of("commitLogDiskRatio", "0.75", "commitLogDiskRatio_/store/a", "0.75",
+ "consumeQueueDiskRatio", "0", "commitLogDirCapacity", "Total : 10 GiB, Free : 2.5 GiB."));
+ Assertions.assertEquals(3, rows(models, RocketmqStorageDiskUsage.class).size());
+ Assertions.assertEquals(2684354560L, rows(models, RocketmqStorageDiskFreeBytes.class).get(0).getValue());
+ var path = rows(models, RocketmqStorageDiskUsage.class).stream().filter(row -> row.getStoreType().equals("commitlog"))
+ .findFirst().orElseThrow();
+ Assertions.assertEquals("/store/a", path.getPath());
+ Assertions.assertEquals(0.75f, path.getValue());
+ for (String value : List.of("-1", "NaN", "Infinity", "1.5", "missing")) {
+ Assertions.assertTrue(collectRuntime(Map.of("commitLogDiskRatio", value)).isEmpty());
+ }
+ for (String value : List.of("Total : 1 GiB, Free : 2 GiB.", "Total : 0 B, Free : 0 B.", "unknown")) {
+ Assertions.assertTrue(collectRuntime(Map.of("commitLogDirCapacity", value)).isEmpty());
+ }
+ Assertions.assertEquals(0L, rows(collectRuntime(Map.of("commitLogDirCapacity", "Total : 1 GiB, Free : 0 B.")),
+ RocketmqStorageDiskFreeBytes.class).get(0).getValue());
+ }
+
+ /** 验证逐客户端采集成功 TPS、失败 TPS 和处理耗时，并将缺失或非法字段补零。 */
+ @Test
+ @DisplayName("逐客户端消费指标采集")
+ public void collectsClientMetricsPerClientAndDefaultsUnavailableFieldsToZero() throws Exception {
+ for (boolean available : List.of(true, false)) {
+ AtomicInteger requests = new AtomicInteger();
+ var models = collectMock((request, callback) -> {
+ switch (request.getCode()) {
+ case RequestCode.GET_ALL_TOPIC_CONFIG -> reply(callback, ResponseCode.SUCCESS,
+ "{\"topicConfigTable\":{\"orders\":{},\"payments\":{}}}".getBytes(StandardCharsets.UTF_8));
+ case RequestCode.QUERY_TOPIC_CONSUME_BY_WHO -> reply(callback, ResponseCode.SUCCESS,
+ "{\"groupList\":[\"buyers\"]}".getBytes(StandardCharsets.UTF_8));
+ case RequestCode.GET_CONSUMER_CONNECTION_LIST -> reply(callback, ResponseCode.SUCCESS,
+ "{\"connectionSet\":[{\"clientId\":\"a\"},{\"clientId\":\"b\"}]}".getBytes(StandardCharsets.UTF_8));
+ case RequestCode.GET_CONSUMER_RUNNING_INFO -> {
+ requests.incrementAndGet();
+ if (available) {
+ reply(callback, ResponseCode.SUCCESS, RemotingSerializable.encode(Map.of("statusTable", Map.of(
+ "orders", Map.of("consumeOKTPS", 2.5, "consumeFailedTPS", 0, "consumeRT", 12.25),
+ "payments", Map.of("consumeOKTPS", -1, "consumeRT", "NaN")))));
+ } else {
+ callback.operationFail(new IllegalStateException("Client disconnected"));
+ callback.operationFail(new IllegalStateException("Duplicate failure notification"));
+ }
+ }
+ default -> callback.operationFail(new IllegalStateException("Not needed for this case"));
+ }
+ });
+ Assertions.assertEquals(2, requests.get(), "Shared group is queried once; each client has one request");
+ if (available) {
+ var success = rows(models, RocketmqConsumerSuccessTps.class);
+ Assertions.assertEquals(Set.of("a", "b"), success.stream().map(RocketmqConsumerSuccessTps::getClientId)
+ .collect(java.util.stream.Collectors.toSet()));
+ Assertions.assertEquals(4, success.size());
+ Assertions.assertTrue(success.stream().allMatch(row ->
+ row.getValue() == (row.getTopicName().equals("orders") ? 2.5f : 0f)));
+ var failed = rows(models, RocketmqConsumerFailedTps.class);
+ Assertions.assertEquals(4, failed.size());
+ Assertions.assertTrue(failed.stream().allMatch(row -> row.getValue() == 0f));
+ var elapsed = rows(models, RocketmqConsumerProcessTime.class);
+ Assertions.assertEquals(4, elapsed.size());
+ Assertions.assertTrue(elapsed.stream().allMatch(row ->
+ row.getValue() == (row.getTopicName().equals("orders") ? 12.25f : 0f)));
+ } else {
+ Assertions.assertFalse(models.containsKey(RocketmqConsumerSuccessTps.class));
+ }
+ }
+ }
+
+ private Map<Class<?>, List<Object>> collectStatsBody(String body) throws Exception {
+ return collectMock((request, callback) -> {
+ if (request.getCode() == RequestCode.GET_BROKER_CONFIG) {
+ reply(callback, ResponseCode.SUCCESS, "brokerClusterName=cluster-A\n".getBytes(StandardCharsets.UTF_8));
+ } else if (request.getCode() == RequestCode.VIEW_BROKER_STATS_DATA) {
+ reply(callback, ResponseCode.SUCCESS, body.getBytes(StandardCharsets.UTF_8));
+ } else {
+ callback.operationFail(new IllegalStateException("Other root unavailable"));
+ }
+ });
+ }
+
+ private Map<Class<?>, List<Object>> collectRuntime(Map<String, String> values) throws Exception {
+ return collectMock((request, callback) -> {
+ if (request.getCode() == RequestCode.GET_BROKER_RUNTIME_INFO) {
+ reply(callback, ResponseCode.SUCCESS, runtimeBody(values));
+ } else {
+ callback.operationFail(new IllegalStateException("Other root unavailable"));
+ }
+ });
+ }
+
+ private byte[] runtimeBody(Map<String, String> values) {
+ KVTable table = new KVTable();
+ table.getTable().putAll(values);
+ return RemotingSerializable.encode(table);
+ }
+
+ private Map<Class<?>, List<Object>> collectMock(java.util.function.BiConsumer<RemotingCommand, InvokeCallback> answer) throws Exception {
+ DefaultRemotingClient client = Mockito.mock(DefaultRemotingClient.class);
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ RuntimeMetadata runtime = runtime();
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(client);
+ Mockito.doAnswer(invocation -> {
+ answer.accept(invocation.getArgument(0), invocation.getArgument(2));
+ return null;
+ }).when(client).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ collector(runtime).collect(0, wrapper);
+ }
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper).sync(capture.capture());
+ return capture.getValue().getDataMap();
+ }
+
+ private <T> List<T> rows(Map<Class<?>, List<Object>> models, Class<T> type) {
+ Assertions.assertTrue(models.containsKey(type), "Missing collected model " + type.getSimpleName());
+ return models.get(type).stream().map(type::cast).toList();
+ }
+
+ private String brokerWindow(Object row) {
+ return row instanceof RocketmqBrokerMessagesIn
+ ? ((RocketmqBrokerMessagesIn) row).getWindow() : ((RocketmqBrokerMessagesOut) row).getWindow();
+ }
+
+ private Long brokerWindowCount(Object row) {
+ return row instanceof RocketmqBrokerMessagesIn
+ ? ((RocketmqBrokerMessagesIn) row).getValueWindowCount() : ((RocketmqBrokerMessagesOut) row).getValueWindowCount();
+ }
+
+ /** 连接真实 Broker，验证原生请求、异步回调和采集模型；需开启 rocketmq.collect.live。 */
+ @Test
+ @DisplayName("真实 Broker 请求与回调采集")
+ public void collectsFromRealBrokerThroughCallbacks() throws Exception {
+ Assumptions.assumeTrue(Boolean.getBoolean("rocketmq.collect.live"));
+ collectFromRealBroker(false);
+ }
+
+ /** 连接真实 Broker 采集指标，写入独立测试库并从 IoTDB 回读；需开启 rocketmq.collect.iotdb.live。 */
+ @Test
+ @DisplayName("真实 Broker 采集及 IoTDB 回读")
+ public void collectsFromRealBrokerAndReadsBackIotdbTables() throws Exception {
+ Assumptions.assumeTrue(Boolean.getBoolean("rocketmq.collect.iotdb.live"));
+ collectFromRealBroker(true);
+ }
+
+ private void collectFromRealBroker(boolean verifyIotdb) throws Exception {
+ RuntimeMetadata runtime = runtime();
+ AbstractSimpleCreateSDKConfig config = ConfigManage.getInstance()
+ .getSimpleCreateSdkConfig(runtime.getClusterType(), SDKTypeEnum.ADMIN);
+ NetAddress address = new NetAddress();
+ address.setAddress("127.0.0.1");
+ address.setPort(Integer.getInteger("rocketmq.collect.port", 21911));
+ config.setNetAddress(address);
+ DefaultRemotingClient real = SDKManage.getInstance().createClient(SDKTypeEnum.ADMIN, runtime, config, runtime.getClusterType());
+ DefaultRemotingClient observed = Mockito.mock(DefaultRemotingClient.class);
+ AtomicInteger outstanding = new AtomicInteger();
+ Set<Integer> successful = ConcurrentHashMap.newKeySet();
+ Map<String, byte[]> brokerResponses = new ConcurrentHashMap<>();
+ var runtimeBefore = new java.util.concurrent.atomic.AtomicLong();
+ var runtimeAfter = new java.util.concurrent.atomic.AtomicLong();
+ Mockito.doAnswer(call -> {
+ RemotingCommand request = call.getArgument(0);
+ InvokeCallback callback = call.getArgument(2);
+ outstanding.incrementAndGet();
+ try {
+ real.invokeAsync(request, call.getArgument(1), new InvokeCallback() {
+ @Override
+ public void operationComplete(ResponseFuture future) {
+ try {
+ if (future.getResponseCommand() != null && future.getResponseCommand().getCode() == ResponseCode.SUCCESS) {
+ successful.add(request.getCode());
+ if (request.getCode() == RequestCode.VIEW_BROKER_STATS_DATA) {
+ var header = (ViewBrokerStatsDataRequestHeader) request.readCustomHeader();
+ brokerResponses.put(header.getStatsName() + ":" + header.getStatsKey(), future.getResponseCommand().getBody());
+ if (header.getStatsName().startsWith("BROKER_")) {
+ brokerResponses.put(header.getStatsName(), future.getResponseCommand().getBody());
+ }
+ } else if (request.getCode() == RequestCode.GET_BROKER_CONFIG
+ || request.getCode() == RequestCode.GET_BROKER_RUNTIME_INFO
+ || request.getCode() == RequestCode.GET_ALL_TOPIC_CONFIG
+ || request.getCode() == RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG) {
+ brokerResponses.put(Integer.toString(request.getCode()), future.getResponseCommand().getBody());
+ }
+ }
+ if (request.getCode() == RequestCode.GET_BROKER_RUNTIME_INFO) {
+ runtimeBefore.set(System.currentTimeMillis());
+ }
+ callback.operationComplete(future);
+ } finally {
+ if (request.getCode() == RequestCode.GET_BROKER_RUNTIME_INFO) {
+ runtimeAfter.set(System.currentTimeMillis());
+ }
+ outstanding.decrementAndGet();
+ }
+ }
+
+ @Override
+ public void operationFail(Throwable error) {
+ callback.operationFail(error);
+ }
+ });
+ } catch (Exception e) {
+ outstanding.decrementAndGet();
+ throw e;
+ }
+ return null;
+ }).when(observed).invokeAsync(Mockito.any(), Mockito.anyLong(), Mockito.any());
+ String topic = "callback_collect_" + java.util.UUID.randomUUID().toString().replace("-", "");
+ SDKManage sdk = Mockito.mock(SDKManage.class);
+ Mockito.when(sdk.getClient(SDKTypeEnum.ADMIN, runtime.getUnique())).thenReturn(observed);
+ try (var mocked = Mockito.mockStatic(SDKManage.class)) {
+ mocked.when(SDKManage::getInstance).thenReturn(sdk);
+ prepareBroker(real, topic);
+ RocketMQCollect collector = collector(runtime);
+ var wrapper = Mockito.mock(org.apache.eventmesh.dashboard.console.function.report.collect.DataSyncHandler.DataSyncHandlerWrapper.class);
+ collector.collect(0, wrapper);
+ var capture = org.mockito.ArgumentCaptor.forClass(org.apache.eventmesh.dashboard.console.function.report.collect.RestoreData.class);
+ Mockito.verify(wrapper).sync(capture.capture());
+ var models = capture.getValue().getDataMap();
+ Assertions.assertEquals(Set.of(
+ org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class,
+ RocketmqConsumerOffset.class, RocketmqConsumerConnectionNumber.class, RocketmqBrokerMessagesIn.class,
+ RocketmqBrokerMessagesOut.class, RocketmqStorageDispatchBehindBytes.class, RocketmqStorageFlushBehindBytes.class,
+ RocketmqStorageMessageReserveTime.class, RocketmqThreadPoolWartermark.class, RocketmqTopicNumber.class,
+ RocketmqConsumerGroupNumber.class, RocketmqTopicMessagesIn.class, RocketmqGroupMessagesOut.class, RocketmqConsumerLagLatency.class, RocketmqConsumerSuccessTps.class, RocketmqConsumerFailedTps.class, RocketmqConsumerProcessTime.class, RocketmqStorageDiskUsage.class, RocketmqStorageDiskFreeBytes.class), models.keySet(),
+ "Only existing Topic/consumer measurements and approved Broker stats/runtime measurements are collected");
+ var requests = org.mockito.ArgumentCaptor.forClass(RemotingCommand.class);
+ Mockito.verify(observed, Mockito.atLeastOnce()).invokeAsync(requests.capture(), Mockito.anyLong(), Mockito.any());
+ Assertions.assertEquals(1L, requests.getAllValues().stream()
+ .filter(request -> request.getCode() == RequestCode.GET_ALL_TOPIC_CONFIG).count(), "Topic list is shared");
+ for (RemotingCommand request : requests.getAllValues()) {
+ Assertions.assertTrue(Set.of(RequestCode.GET_ALL_TOPIC_CONFIG, RequestCode.GET_TOPIC_STATS_INFO,
+ RequestCode.QUERY_TOPIC_CONSUME_BY_WHO, RequestCode.GET_CONSUMER_CONNECTION_LIST,
+ RequestCode.GET_CONSUME_STATS, RequestCode.GET_BROKER_CONFIG, RequestCode.VIEW_BROKER_STATS_DATA,
+ RequestCode.GET_BROKER_RUNTIME_INFO, RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG,
+ RequestCode.QUERY_CONSUME_TIME_SPAN, RequestCode.GET_CONSUMER_RUNNING_INFO).contains(request.getCode()), "Unexpected collection family");
+ if (request.getCode() == RequestCode.GET_CONSUME_STATS) {
+ var header = (org.apache.rocketmq.remoting.protocol.header.GetConsumeStatsRequestHeader) request.readCustomHeader();
+ Assertions.assertNotNull(header.getTopic(), "Consumer query must be scoped to its Topic");
+ }
+ }
+ var positions = models.get(org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset.class);
+ Assertions.assertTrue(positions.stream().map(row ->
+ (org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.Rocketmq2ProducerOffset) row)
+ .anyMatch(row -> topic.equals(row.getTopicName()) && Long.valueOf(3).equals(row.getValue())));
+ Assertions.assertTrue(models.containsKey(RocketmqConsumerConnectionNumber.class));
+ var connectionRows = models.get(RocketmqConsumerConnectionNumber.class);
+ Assertions.assertTrue(connectionRows.stream().map(RocketmqConsumerConnectionNumber.class::cast)
+ .anyMatch(row -> topic.equals(row.getGroupName()) && Long.valueOf(1L).equals(row.getValueConnectionCount())));
+ var consumers = models.get(org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerOffset.class);
+ Assertions.assertTrue(consumers.stream().map(row ->
+ (org.apache.eventmesh.dashboard.console.function.report.model.rocketmq.RocketmqConsumerOffset) row)
+ .anyMatch(row -> topic.equals(row.getTopicName()) && topic.equals(row.getGroupName())
+ && Long.valueOf(2).equals(row.getValueConsumerOffset()) && Long.valueOf(1).equals(row.getValueOffsetLag())));
+ consumers.stream().map(RocketmqConsumerOffset.class::cast).filter(row -> topic.equals(row.getTopicName()))
+ .forEach(row -> LoggerFactory.getLogger(RocketMQCollectTest.class).info(
+ "消费进度验证：Topic={} Group={} Queue={} consumerOffset={} brokerOffset={} lag={}",
+ row.getTopicName(), row.getGroupName(), row.getQueueId(), row.getValueConsumerOffset(),
+ row.getValueBrokerOffset(), row.getValueOffsetLag()));
+ long deadline = System.currentTimeMillis() + 15000;
+ while (outstanding.get() != 0 && System.currentTimeMillis() < deadline) {
+ Thread.sleep(20);
+ }
+ Assertions.assertEquals(0, outstanding.get(), "All callbacks must finish");
+ for (int code : List.of(RequestCode.GET_ALL_TOPIC_CONFIG, RequestCode.GET_TOPIC_STATS_INFO,
+ RequestCode.QUERY_TOPIC_CONSUME_BY_WHO, RequestCode.GET_CONSUME_STATS, RequestCode.GET_CONSUMER_CONNECTION_LIST,
+ RequestCode.GET_BROKER_CONFIG, RequestCode.GET_BROKER_RUNTIME_INFO, RequestCode.VIEW_BROKER_STATS_DATA,
+ RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG)) {
+ Assertions.assertTrue(successful.contains(code), "Missing successful request " + code);
+ }
+ assertLiveBrokerModels(models, runtime, brokerResponses, runtimeBefore.get(), runtimeAfter.get());
+ assertScopeModels(models, brokerResponses, topic);
+ Assertions.assertTrue(successful.contains(RequestCode.GET_CONSUMER_RUNNING_INFO));
+ Assertions.assertTrue(successful.contains(RequestCode.QUERY_CONSUME_TIME_SPAN));
+ var successfulClient = rows(models, RocketmqConsumerSuccessTps.class).stream()
+ .filter(row -> row.getTopicName().equals(topic) && row.getClientId().equals(topic)).findFirst().orElseThrow();
+ Assertions.assertEquals(topic, successfulClient.getGroupName());
+ Assertions.assertEquals(2.5f, successfulClient.getValue());
+ Assertions.assertEquals(0.5f, rows(models, RocketmqConsumerFailedTps.class).stream()
+ .filter(row -> row.getTopicName().equals(topic)).findFirst().orElseThrow().getValue());
+ Assertions.assertEquals(12.25f, rows(models, RocketmqConsumerProcessTime.class).stream()
+ .filter(row -> row.getTopicName().equals(topic)).findFirst().orElseThrow().getValue());
+ var lag = rows(models, RocketmqConsumerLagLatency.class).stream()
+ .filter(row -> row.getTopicName().equals(topic)).findFirst().orElseThrow();
+ Assertions.assertEquals("0", lag.getQueueId());
+ Assertions.assertTrue(lag.getValue() >= 0 && lag.getValue() < 30000);
+ var nativeRuntime = (Map<?, ?>) RemotingSerializable.decode(
+ brokerResponses.get(Integer.toString(RequestCode.GET_BROKER_RUNTIME_INFO)), Map.class).get("table");
+ for (var disk : rows(models, RocketmqStorageDiskUsage.class)) {
+ String key = disk.getStoreType().equals("commitlog") ? "commitLogDiskRatio_" + disk.getPath()
+ : disk.getStoreType().equals("commitlog_min") ? "commitLogDiskRatio" : "consumeQueueDiskRatio";
+ Assertions.assertEquals(Float.parseFloat(nativeRuntime.get(key).toString()), disk.getValue());
+ }
+
+ var nativeConfig = new java.util.Properties();
+ nativeConfig.load(new java.io.StringReader(new String(
+ brokerResponses.get(Integer.toString(RequestCode.GET_BROKER_CONFIG)), StandardCharsets.UTF_8)));
+ for (int code : List.of(RequestCode.GET_BROKER_CONFIG, RequestCode.GET_BROKER_RUNTIME_INFO)) {
+ Assertions.assertEquals(1L, requests.getAllValues().stream().filter(request -> request.getCode() == code).count());
+ }
+ var statsRequests = requests.getAllValues().stream()
+ .filter(request -> request.getCode() == RequestCode.VIEW_BROKER_STATS_DATA)
+ .filter(request -> ((ViewBrokerStatsDataRequestHeader) request.readCustomHeader()).getStatsName().startsWith("BROKER_"))
+ .toList();
+ Assertions.assertEquals(2, statsRequests.size());
+ Set<String> names = new java.util.HashSet<>();
+ for (RemotingCommand request : statsRequests) {
+ var header = (ViewBrokerStatsDataRequestHeader) request.readCustomHeader();
+ Assertions.assertEquals(nativeConfig.getProperty("brokerClusterName"), header.getStatsKey());
+ names.add(header.getStatsName());
+ }
+ Assertions.assertEquals(Set.of("BROKER_PUT_NUMS", "BROKER_GET_NUMS"), names);
+ if (verifyIotdb) {
+ verifyIotdbTables(models, runtime, topic);
+ }
+ } finally {
+ try {
+ DeleteTopicRequestHeader delete = new DeleteTopicRequestHeader();
+ delete.setTopic(topic);
+ assertSuccess(real.invokeSync(RemotingCommand.createRequestCommand(RequestCode.DELETE_TOPIC_IN_BROKER, delete), 3000));
+ DeleteSubscriptionGroupRequestHeader deleteGroup = new DeleteSubscriptionGroupRequestHeader();
+ deleteGroup.setGroupName(topic);
+ assertSuccess(real.invokeSync(RemotingCommand.createRequestCommand(RequestCode.DELETE_SUBSCRIPTIONGROUP, deleteGroup), 3000));
+ } finally {
+ real.shutdown();
+ SDKManage.getInstance().deleteClient(null, runtime.getUnique());
+ }
+ }
+ }
+
+ /** 验证配置中的 Topic、消费组总数，以及按 Topic 和消费组区分的消息窗口统计。 */
+ @Test
+ @DisplayName("资源总数与分范围消息统计")
+ public void countsConfiguredResourcesAndScopesMessageStats() throws Exception {
+ Set<String> statsKeys = new java.util.HashSet<>();
+ AtomicInteger groupRequests = new AtomicInteger();
+ var models = collectMock((request, callback) -> {
+ switch (request.getCode()) {
+ case RequestCode.GET_ALL_TOPIC_CONFIG:
+ reply(callback, ResponseCode.SUCCESS,
+ "{\"topicConfigTable\":{\"orders\":{},\"payments\":{},\"%DLQ%buyers\":{}}}".getBytes(StandardCharsets.UTF_8));
+ break;
+ case RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG:
+ groupRequests.incrementAndGet();
+ reply(callback, ResponseCode.SUCCESS,
+ "{\"subscriptionGroupTable\":{\"buyers\":{},\"offline\":{},\"system\":{}}}".getBytes(StandardCharsets.UTF_8));
+ break;
+ case RequestCode.QUERY_TOPIC_CONSUME_BY_WHO:
+ reply(callback, ResponseCode.SUCCESS, "{\"groupList\":[\"buyers\"]}".getBytes(StandardCharsets.UTF_8));
+ break;
+ case RequestCode.VIEW_BROKER_STATS_DATA:
+ var header = (ViewBrokerStatsDataRequestHeader) request.readCustomHeader();
+ Assertions.assertTrue(statsKeys.add(header.getStatsName() + ":" + header.getStatsKey()));
+ reply(callback, ResponseCode.SUCCESS,
+ ("{\"statsMinute\":{\"sum\":9007199254740993,\"tps\":1.5},"
+ + "\"statsHour\":{\"sum\":0,\"tps\":0},\"statsDay\":{\"sum\":-1,\"tps\":1}}")
+ .getBytes(StandardCharsets.UTF_8));
+ callback.operationFail(new IllegalStateException("duplicate"));
+ break;
+ default:
+ callback.operationFail(new IllegalStateException("Unrelated data unavailable"));
+ }
+ });
+ Assertions.assertEquals(3L, rows(models, RocketmqTopicNumber.class).get(0).getValue());
+ Assertions.assertEquals(3L, rows(models, RocketmqConsumerGroupNumber.class).get(0).getValue());
+ Assertions.assertEquals(1, groupRequests.get());
+ Assertions.assertEquals(Set.of("TOPIC_PUT_NUMS:orders", "TOPIC_PUT_NUMS:payments", "TOPIC_PUT_NUMS:%DLQ%buyers",
+ "GROUP_GET_NUMS:orders@buyers", "GROUP_GET_NUMS:payments@buyers"), statsKeys);
+ Assertions.assertEquals(6, rows(models, RocketmqTopicMessagesIn.class).size());
+ Assertions.assertEquals(4, rows(models, RocketmqGroupMessagesOut.class).size());
+ rows(models, RocketmqGroupMessagesOut.class).forEach(row -> {
+ Assertions.assertEquals("buyers", row.getGroupName());
+ Assertions.assertEquals("minute".equals(row.getWindow()) ? 9007199254740993L : 0L, row.getValueWindowCount());
+ Assertions.assertEquals("minute".equals(row.getWindow()) ? 1.5F : 0F, row.getValue());
+ Assertions.assertNotNull(row.getRuntimeId());
+ });
+ }
+
+ /** 验证配置查询失败不生成零值，成功返回空配置时生成零值。 */
+ @Test
+ @DisplayName("缺失配置与空配置的零值语义")
+ public void missingConfigurationIsNotZeroButEmptyConfigurationIsZero() throws Exception {
+ for (String body : List.of("{}", "{\"topicConfigTable\":null,\"subscriptionGroupTable\":null}", "not-json")) {
+ var models = collectMock((request, callback) -> {
+ if (request.getCode() == RequestCode.GET_ALL_TOPIC_CONFIG || request.getCode() == RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG) {
+ reply(callback, ResponseCode.SUCCESS, body.getBytes(StandardCharsets.UTF_8));
+ } else {
+ callback.operationFail(new IllegalStateException("Unavailable"));
+ }
+ });
+ Assertions.assertTrue(models.isEmpty(), body);
+ }
+ var empty = collectMock((request, callback) -> {
+ if (request.getCode() == RequestCode.GET_ALL_TOPIC_CONFIG || request.getCode() == RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG) {
+ reply(callback, ResponseCode.SUCCESS,
+ "{\"topicConfigTable\":{},\"subscriptionGroupTable\":{}}".getBytes(StandardCharsets.UTF_8));
+ } else {
+ callback.operationFail(new IllegalStateException("Unavailable"));
+ }
+ });
+ Assertions.assertEquals(0L, rows(empty, RocketmqTopicNumber.class).get(0).getValue());
+ Assertions.assertEquals(0L, rows(empty, RocketmqConsumerGroupNumber.class).get(0).getValue());
+ }
+
+ /** 验证 Topic 和消费组指标使用名称与窗口标签，并保持 GAUGE 语义。 */
+ @Test
+ @DisplayName("Topic 与消费组指标表结构")
+ public void scopedModelsHaveNameAndWindowTagsAndGaugeSemantics() {
+ assertReportTable(RocketmqTopicNumber.class, "rocketmq_topic_number", List.of("runtime_id"), List.of("value"));
+ assertReportTable(RocketmqConsumerGroupNumber.class, "rocketmq_consumer_group_number", List.of("runtime_id"), List.of("value"));
+ assertReportTable(RocketmqTopicMessagesIn.class, "rocketmq_topic_messages_in",
+ List.of("topic_name", "window"), List.of("value_window_count"), List.of("value"));
+ assertReportTable(RocketmqGroupMessagesOut.class, "rocketmq_group_messages_out",
+ List.of("topic_name", "group_name", "window"), List.of("value_window_count"), List.of("value"));
+ var row = RocketMQCollectMapper.INSTANCE.groupMessages("orders", "buyers", "minute", 9007199254740993L, 1.5F);
+ Assertions.assertEquals("orders", row.getTopicName());
+ Assertions.assertEquals("buyers", row.getGroupName());
+ Assertions.assertEquals(9007199254740993L, row.getValueWindowCount());
+ Assertions.assertEquals(1.5F, row.getValue());
+ }
+
+ private void assertScopeModels(Map<Class<?>, List<Object>> models, Map<String, byte[]> responses, String topic) {
+ Map<?, ?> topics = RemotingSerializable.decode(responses.get(Integer.toString(RequestCode.GET_ALL_TOPIC_CONFIG)), Map.class);
+ Map<?, ?> groups = RemotingSerializable.decode(responses.get(Integer.toString(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG)), Map.class);
+ Assertions.assertEquals((long) ((Map<?, ?>) topics.get("topicConfigTable")).size(), rows(models, RocketmqTopicNumber.class).get(0).getValue());
+ Assertions.assertEquals((long) ((Map<?, ?>) groups.get("subscriptionGroupTable")).size(),
+ rows(models, RocketmqConsumerGroupNumber.class).get(0).getValue());
+ var incoming = rows(models, RocketmqTopicMessagesIn.class).stream().filter(row -> topic.equals(row.getTopicName())).toList();
+ var outgoing = rows(models, RocketmqGroupMessagesOut.class).stream().filter(row -> topic.equals(row.getTopicName())
+ && topic.equals(row.getGroupName())).toList();
+ Assertions.assertEquals(3, incoming.size());
+ Assertions.assertEquals(3, outgoing.size());
+ for (Object row : java.util.stream.Stream.concat(incoming.stream(), outgoing.stream()).toList()) {
+ boolean in = row instanceof RocketmqTopicMessagesIn;
+ String window = in ? ((RocketmqTopicMessagesIn) row).getWindow() : ((RocketmqGroupMessagesOut) row).getWindow();
+ long count = in ? ((RocketmqTopicMessagesIn) row).getValueWindowCount() : ((RocketmqGroupMessagesOut) row).getValueWindowCount();
+ String key = in ? "TOPIC_PUT_NUMS:" + topic : "GROUP_GET_NUMS:" + topic + "@" + topic;
+ Map<?, ?> wire = RemotingSerializable.decode(responses.get(key), Map.class);
+ Map<?, ?> sample = (Map<?, ?>) wire.get(Map.of("minute", "statsMinute", "hour", "statsHour", "day", "statsDay").get(window));
+ Assertions.assertEquals(Long.parseLong(sample.get("sum").toString()), count);
+ Assertions.assertEquals(((Number) sample.get("tps")).floatValue(), ((RuntimeFloatValue) row).getValue());
+ }
+ }
+
+ private void assertLiveBrokerModels(Map<Class<?>, List<Object>> models, RuntimeMetadata runtime,
+ Map<String, byte[]> responses, long before, long after) {
+ Map<String, String> windows = Map.of("minute", "statsMinute", "hour", "statsHour", "day", "statsDay");
+ Map<Class<?>, String> stats = Map.of(RocketmqBrokerMessagesIn.class, "BROKER_PUT_NUMS",
+ RocketmqBrokerMessagesOut.class, "BROKER_GET_NUMS");
+ for (var entry : stats.entrySet()) {
+ byte[] body = responses.get(entry.getValue());
+ Assertions.assertNotNull(body, entry.getValue());
+ var wire = RemotingSerializable.decode(body, java.util.HashMap.class);
+ List<Object> samples = models.get(entry.getKey());
+ Assertions.assertEquals(3, samples.size(), "Native Broker responses expose all three windows, including valid zero windows");
+ Set<String> observedWindows = new java.util.HashSet<>();
+ for (Object sample : samples) {
+ String name = brokerWindow(sample);
+ Assertions.assertTrue(observedWindows.add(name));
+ Assertions.assertInstanceOf(Map.class, wire.get(windows.get(name)));
+ var nativeWindow = (Map<?, ?>) wire.get(windows.get(name));
+ Assertions.assertInstanceOf(Number.class, nativeWindow.get("sum"));
+ Assertions.assertInstanceOf(Number.class, nativeWindow.get("tps"));
+ long count = new java.math.BigDecimal(nativeWindow.get("sum").toString()).longValueExact();
+ float rate = ((Number) nativeWindow.get("tps")).floatValue();
+ Assertions.assertTrue(count >= 0 && rate >= 0 && Float.isFinite(rate));
+ Assertions.assertEquals(count, brokerWindowCount(sample));
+ Assertions.assertEquals(Float.valueOf(rate), ((RuntimeFloatValue) sample).getValue());
+ Assertions.assertEquals(runtime.getId(), ((RuntimeFloatValue) sample).getRuntimeId());
+ LoggerFactory.getLogger(RocketMQCollectTest.class).info(
+ "Broker 窗口验证：stats={} window={} count={} messagesPerSecond={}", entry.getValue(), name, count, rate);
+ }
+ Assertions.assertEquals(windows.keySet(), observedWindows);
+ }
+ byte[] body = responses.get(Integer.toString(RequestCode.GET_BROKER_RUNTIME_INFO));
+ Assertions.assertNotNull(body);
+ var values = RemotingSerializable.decode(body, KVTable.class).getTable();
+ long dispatch = Long.parseLong(values.get("dispatchBehindBytes"));
+ Assertions.assertEquals(1, rows(models, RocketmqStorageDispatchBehindBytes.class).size());
+ Assertions.assertEquals(dispatch, rows(models, RocketmqStorageDispatchBehindBytes.class).get(0).getValue());
+ long flush = expectedNativeBytes(values.get("remainHowManyDataToFlush"));
+ if (values.containsKey("remainHowManyDataToCommit")) {
+ flush = Math.addExact(flush, expectedNativeBytes(values.get("remainHowManyDataToCommit")));
+ }
+ Assertions.assertEquals(1, rows(models, RocketmqStorageFlushBehindBytes.class).size());
+ Assertions.assertEquals(flush, rows(models, RocketmqStorageFlushBehindBytes.class).get(0).getValue());
+ long earliest = Long.parseLong(values.get("earliestMessageTimeStamp"));
+ Assertions.assertTrue(earliest > 0 && earliest <= before, "Fixture must expose a real earliest stored message");
+ Assertions.assertEquals(1, rows(models, RocketmqStorageMessageReserveTime.class).size());
+ long age = rows(models, RocketmqStorageMessageReserveTime.class).get(0).getValue();
+ Assertions.assertTrue(age >= before - earliest && age <= after - earliest, "Native timestamp is converted to age in ms");
+ Map<String, String> poolKeys = Map.of("send", "sendThreadPoolQueueSize", "pull", "pullThreadPoolQueueSize",
+ "litePull", "litePullThreadPoolQueueSize", "query", "queryThreadPoolQueueSize", "ack", "ackThreadPoolQueueSize",
+ "endTransaction", "EndTransactionQueueSize");
+ Map<String, Long> expectedPools = new java.util.HashMap<>();
+ poolKeys.forEach((pool, key) -> {
+ if (values.containsKey(key)) {
+ expectedPools.put(pool, Long.parseLong(values.get(key)));
+ }
+ });
+ Assertions.assertFalse(expectedPools.isEmpty(), "Fixture must expose native thread pool queues");
+ Map<String, Long> actualPools = new java.util.HashMap<>();
+ rows(models, RocketmqThreadPoolWartermark.class).forEach(row ->
+ Assertions.assertNull(actualPools.put(row.getPoolName(), row.getValue()), "Duplicate pool sample"));
+ Assertions.assertEquals(expectedPools, actualPools);
+ LoggerFactory.getLogger(RocketMQCollectTest.class).info(
+ "Broker 运行时验证：dispatchBytes={} flushAndCommitBytes={} reserveAgeMs={} poolQueues={}", dispatch, flush, age, actualPools);
+ }
+
+ private long expectedNativeBytes(String value) {
+ Assertions.assertNotNull(value, "Native byte value must be present");
+ var matcher = java.util.regex.Pattern.compile("([0-9]+(?:\\.[0-9]+)?)\\s*(B|KiB|MiB|GiB|TiB|PiB|EiB)?")
+ .matcher(value.trim());
+ Assertions.assertTrue(matcher.matches(), value);
+ String unit = matcher.group(2);
+ int power = unit == null || "B".equals(unit) ? 0 : List.of("KiB", "MiB", "GiB", "TiB", "PiB", "EiB").indexOf(unit) + 1;
+ return new java.math.BigDecimal(matcher.group(1)).multiply(java.math.BigDecimal.valueOf(1024).pow(power))
+ .setScale(0, java.math.RoundingMode.HALF_UP).longValueExact();
+ }
+
+ /** 使用真实报表引擎建表和写入，再通过独立 JDBC 连接读回本次采集的数据。 */
+ private void verifyIotdbTables(java.util.Map<Class<?>, List<Object>> models, RuntimeMetadata runtime, String topic) throws Exception {
+ var engine = new org.apache.eventmesh.dashboard.console.function.report.iotdb.IotDBReportEngine();
+ Class.forName("org.apache.iotdb.jdbc.IoTDBDriver");
+ String address = System.getProperty("rocketmq.collect.iotdb.address", "127.0.0.1:6667");
+ String configuredDatabase = System.getProperty("rocketmq.collect.iotdb.database");
+ String database = configuredDatabase == null
+ ? "collect_test_" + java.util.UUID.randomUUID().toString().replace("-", "") : configuredDatabase;
+ Assertions.assertTrue(database.matches("collect_test_[a-z0-9_]+"), "Use a dedicated collection test database");
+ String url = "jdbc:iotdb://" + address + "/" + database + "?sql_dialect=table";
+ var tables = new java.util.HashMap<Class<?>, String>(java.util.Map.<Class<?>, String>of(
+ RocketmqConsumerOffset.class, "rocketmq_consumer_offset",
+ RocketmqConsumerConnectionNumber.class, "rocketmq_consumer_connection_number",
+ RocketmqBrokerMessagesIn.class, "rocketmq_broker_messages_in",
+ RocketmqBrokerMessagesOut.class, "rocketmq_broker_messages_out",
+ RocketmqStorageDispatchBehindBytes.class, "rocketmq_storage_dispatch_behind_bytes",
+ RocketmqStorageFlushBehindBytes.class, "rocketmq_storage_flush_behind_bytes",
+ RocketmqStorageMessageReserveTime.class, "rocketmq_storage_message_reserve_time",
+ RocketmqThreadPoolWartermark.class, "rocketmq_thread_pool_wartermark"));
+ tables.put(RocketmqTopicNumber.class, "rocketmq_topic_number");
+ tables.put(RocketmqConsumerGroupNumber.class, "rocketmq_consumer_group_number");
+ tables.put(RocketmqTopicMessagesIn.class, "rocketmq_topic_messages_in");
+ tables.put(RocketmqGroupMessagesOut.class, "rocketmq_group_messages_out");
+ tables.put(RocketmqConsumerLagLatency.class, "rocketmq_consumer_lag_latency");
+ tables.put(RocketmqConsumerSuccessTps.class, "rocketmq_consumer_success_tps");
+ tables.put(RocketmqConsumerFailedTps.class, "rocketmq_consumer_failed_tps");
+ tables.put(RocketmqConsumerProcessTime.class, "rocketmq_consumer_process_time");
+ tables.put(RocketmqStorageDiskUsage.class, "rocketmq_storage_disk_usage");
+ tables.put(RocketmqStorageDiskFreeBytes.class, "rocketmq_storage_disk_free_bytes");
+ engine.setClazzToTableName(tables);
+ try {
+ // 使用独立测试库，避免同名历史表结构影响测试；建表及写入仍调用现有报表引擎。
+ try (var connection = java.sql.DriverManager.getConnection(
+ "jdbc:iotdb://" + address + "/?sql_dialect=table", "root", "root");
+ var statement = connection.createStatement()) {
+ statement.execute("create database if not exists " + database);
+ }
+ var source = new com.alibaba.druid.pool.DruidDataSource();
+ org.apache.commons.lang3.reflect.FieldUtils.writeField(engine, "dataSource", source, true);
+ source.setUrl(url);
+ source.setDriverClassName("org.apache.iotdb.jdbc.IoTDBDriver");
+ source.setUsername("root");
+ source.setPassword("root");
+ source.setMaxActive(2);
+ source.setMaxWait(5000);
+ source.init();
+ for (var entry : tables.entrySet()) {
+ Class<?> model = entry.getKey();
+ ReportMeta annotation = model.getAnnotation(ReportMeta.class);
+ var metadata = new ReportMetaData();
+ metadata.setClazz(model);
+ metadata.setTableName(annotation.tableName());
+ metadata.setReportName(annotation.reportName());
+ metadata.setComment(annotation.comment());
+ var fields = org.apache.commons.lang3.reflect.FieldUtils.getAllFieldsList(model);
+ engine.createReportHandler(metadata, fields);
+ engine.createReport(entry.getValue());
+ }
+ var rows = new java.util.HashMap<Class<?>, List<Object>>();
+ rows.put(RocketmqConsumerOffset.class, models.get(RocketmqConsumerOffset.class).stream()
+ .filter(row -> topic.equals(((RocketmqConsumerOffset) row).getTopicName())).toList());
+ rows.put(RocketmqConsumerConnectionNumber.class, models.get(RocketmqConsumerConnectionNumber.class).stream()
+ .filter(row -> topic.equals(((RocketmqConsumerConnectionNumber) row).getGroupName())).toList());
+ Assertions.assertFalse(rows.get(RocketmqConsumerOffset.class).isEmpty());
+ Assertions.assertEquals(1, rows.get(RocketmqConsumerConnectionNumber.class).size());
+ for (Class<?> model : tables.keySet()) {
+ if (model != RocketmqConsumerOffset.class && model != RocketmqConsumerConnectionNumber.class) {
+ Assertions.assertTrue(models.containsKey(model), "Missing live observations for " + model.getSimpleName());
+ Assertions.assertFalse(models.get(model).isEmpty());
+ rows.put(model, models.get(model));
+ }
+ }
+ // Producer aggregate rows have nullable numeric fields; the unchanged formatter is not exercised for those rows.
+ engine.batchInsertByClass(rows);
+ try (var connection = java.sql.DriverManager.getConnection(
+ url, "root", "root");
+ var statement = connection.createStatement()) {
+ for (var entry : tables.entrySet()) {
+ String sql = "select * from " + entry.getValue() + " where runtime_id = '" + runtime.getId() + "'";
+ try (var result = statement.executeQuery(sql)) {
+ int count = 0;
+ boolean expectedSample = false;
+ Set<String> dimensions = new java.util.HashSet<>();
+ while (result.next()) {
+ count++;
+ Assertions.assertEquals("7", result.getString("organization_id"));
+ Assertions.assertEquals(runtime.getClusterId().toString(), result.getString("clusters_id"));
+ Assertions.assertEquals(runtime.getId().toString(), result.getString("runtime_id"));
+ Assertions.assertNotNull(result.getObject("time"));
+ if (entry.getKey() == RocketmqConsumerOffset.class) {
+ Assertions.assertEquals(topic, result.getString("group_name"));
+ Assertions.assertEquals(topic, result.getString("topic_name"));
+ Assertions.assertEquals("0", result.getString("queue_id"));
+ expectedSample |= result.getLong("value_consumer_offset") == 2L
+ && result.getLong("value_broker_offset") == 3L && result.getLong("value_offset_lag") == 1L;
+ } else if (entry.getKey() == RocketmqConsumerConnectionNumber.class) {
+ Assertions.assertEquals(topic, result.getString("group_name"));
+ long expected = ((RocketmqConsumerConnectionNumber) rows.get(entry.getKey()).get(0)).getValueConnectionCount();
+ Assertions.assertEquals(expected, result.getLong("value_connection_count"));
+ expectedSample = expected == 1;
+ } else if (entry.getKey() == RocketmqTopicMessagesIn.class || entry.getKey() == RocketmqGroupMessagesOut.class) {
+ String observedTopic = result.getString("topic_name");
+ String window = result.getString("window");
+ String group = entry.getKey() == RocketmqGroupMessagesOut.class ? result.getString("group_name") : "";
+ Object expected = rows.get(entry.getKey()).stream().filter(row -> {
+ if (row instanceof RocketmqTopicMessagesIn incoming) {
+ return observedTopic.equals(incoming.getTopicName()) && window.equals(incoming.getWindow());
+ }
+ RocketmqGroupMessagesOut outgoing = (RocketmqGroupMessagesOut) row;
+ return observedTopic.equals(outgoing.getTopicName()) && window.equals(outgoing.getWindow())
+ && group.equals(outgoing.getGroupName());
+ }).findFirst().orElseThrow();
+ long countValue = expected instanceof RocketmqTopicMessagesIn incoming
+ ? incoming.getValueWindowCount() : ((RocketmqGroupMessagesOut) expected).getValueWindowCount();
+ Assertions.assertEquals(countValue, result.getLong("value_window_count"));
+ Assertions.assertEquals(((RuntimeFloatValue) expected).getValue(), result.getFloat("value"));
+ Assertions.assertTrue(dimensions.add(observedTopic + ":" + group + ":" + window));
+ expectedSample = true;
+ } else {
+ String dimension = assertBrokerReadback(result, entry.getKey(), rows.get(entry.getKey()));
+ Assertions.assertTrue(dimensions.add(dimension), "Duplicate persisted dimension " + dimension);
+ expectedSample = true;
+ }
+ }
+ Assertions.assertEquals(rows.get(entry.getKey()).size(), count, entry.getValue());
+ Assertions.assertTrue(expectedSample, "Missing expected sample in " + entry.getValue());
+ LoggerFactory.getLogger(RocketMQCollectTest.class).info(
+ "IoTDB readback verified database={} table={} runtime={} topic={} rows={}",
+ database, entry.getValue(), runtime.getId(), topic, count);
+ }
+ }
+ }
+ } finally {
+ var source = (com.alibaba.druid.pool.DruidDataSource)
+ org.apache.commons.lang3.reflect.FieldUtils.readField(engine, "dataSource", true);
+ if (source != null) {
+ source.close();
+ }
+ if (configuredDatabase == null) {
+ try (var connection = java.sql.DriverManager.getConnection(
+ "jdbc:iotdb://" + address + "/?sql_dialect=table", "root", "root");
+ var statement = connection.createStatement()) {
+ statement.execute("drop database if exists " + database);
+ }
+ }
+ }
+ }
+
+ private String assertBrokerReadback(java.sql.ResultSet result, Class<?> model, List<Object> expectedRows) throws Exception {
+ if (Set.of(RocketmqConsumerLagLatency.class, RocketmqConsumerSuccessTps.class, RocketmqConsumerFailedTps.class, RocketmqConsumerProcessTime.class, RocketmqStorageDiskUsage.class, RocketmqStorageDiskFreeBytes.class).contains(model)) {
+ var tags = org.apache.commons.lang3.reflect.FieldUtils.getAllFieldsList(model).stream()
+ .filter(field -> field.isAnnotationPresent(ReportTag.class)).toList();
+ List<Object> matches = new java.util.ArrayList<>();
+ for (Object row : expectedRows) {
+ boolean match = true;
+ for (var tag : tags) {
+ String column = tag.getName().replaceAll("([a-z])([A-Z])", "$1_$2").toLowerCase(java.util.Locale.ROOT);
+ Object value = org.apache.commons.lang3.reflect.FieldUtils.readField(tag, row, true);
+ match &= java.util.Objects.equals(value == null ? null : value.toString(), result.getString(column));
+ }
+ if (match) {
+ matches.add(row);
+ }
+ }
+ Assertions.assertEquals(1, matches.size());
+ Object row = matches.get(0);
+ if (row instanceof RuntimeFloatValue numeric) {
+ Assertions.assertEquals(numeric.getValue(), result.getFloat("value"));
+ } else {
+ Assertions.assertEquals(((RuntimeLongValue) row).getValue(), result.getLong("value"));
+ }
+ Assertions.assertFalse(result.wasNull());
+ return row.toString();
+ }
+ String dimension = "broker";
+ Object expected;
+ if (model == RocketmqBrokerMessagesIn.class || model == RocketmqBrokerMessagesOut.class) {
+ String window = result.getString("window");
+ expected = expectedRows.stream().filter(row -> window.equals(brokerWindow(row))).findFirst().orElseThrow();
+ Assertions.assertEquals(brokerWindowCount(expected).longValue(), result.getLong("value_window_count"));
+ Assertions.assertFalse(result.wasNull());
+ Assertions.assertEquals(((RuntimeFloatValue) expected).getValue().floatValue(), result.getFloat("value"));
+ Assertions.assertFalse(result.wasNull());
+ dimension = window;
+ } else {
+ if (model == RocketmqThreadPoolWartermark.class) {
+ String pool = result.getString("pool_name");
+ expected = expectedRows.stream().map(RocketmqThreadPoolWartermark.class::cast)
+ .filter(row -> pool.equals(row.getPoolName())).findFirst().orElseThrow();
+ dimension = pool;
+ } else {
+ Assertions.assertEquals(1, expectedRows.size());
+ expected = expectedRows.get(0);
+ }
+ Assertions.assertEquals(((RuntimeLongValue) expected).getValue().longValue(), result.getLong("value"));
+ Assertions.assertFalse(result.wasNull());
+ }
+ LoggerFactory.getLogger(RocketMQCollectTest.class).info("IoTDB Broker sample verified model={} dimension={} value={}",
+ model.getSimpleName(), dimension, result.getObject("value"));
+ return dimension;
+ }
+
+ private void assertSuccess(RemotingCommand response) {
+ Assertions.assertEquals(ResponseCode.SUCCESS, response.getCode(), response.getRemark());
+ }
+
+ private void prepareBroker(DefaultRemotingClient client, String topic) throws Exception {
+ // 真实 Broker 转发请求到已注册连接；客户端响应采用固定测试样本，不代表业务压力测试。
+ client.registerProcessor(RequestCode.GET_CONSUMER_RUNNING_INFO, new org.apache.rocketmq.remoting.netty.NettyRequestProcessor() {
+ @Override
+ public RemotingCommand processRequest(io.netty.channel.ChannelHandlerContext context, RemotingCommand request) {
+ RemotingCommand response = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, null);
+ response.setBody(RemotingSerializable.encode(Map.of("statusTable", Map.of(topic,
+ Map.of("consumeOKTPS", 2.5, "consumeFailedTPS", 0.5, "consumeRT", 12.25)))));
+ return response;
+ }
+
+ @Override
+ public boolean rejectRequest() {
+ return false;
+ }
+ }, null);
+ CreateTopicRequestHeader create = new CreateTopicRequestHeader();
+ create.setTopic(topic);
+ create.setDefaultTopic("TBW102");
+ create.setReadQueueNums(1);
+ create.setWriteQueueNums(1);
+ create.setPerm(6);
+ create.setTopicFilterType("SINGLE_TAG");
+ assertSuccess(client.invokeSync(RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_TOPIC, create), 3000));
+ HeartbeatData heartbeat = new HeartbeatData();
+ heartbeat.setClientID(topic);
+ ProducerData producer = new ProducerData();
+ producer.setGroupName(topic);
+ heartbeat.getProducerDataSet().add(producer);
+ RemotingCommand beat = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null);
+ beat.setBody(RemotingSerializable.encode(heartbeat));
+ assertSuccess(client.invokeSync(beat, 3000));
+ for (int i = 0; i < 3; i++) {
+ SendMessageRequestHeader header = new SendMessageRequestHeader();
+ header.setProducerGroup(topic);
+ header.setTopic(topic);
+ header.setDefaultTopic(topic);
+ header.setDefaultTopicQueueNums(1);
+ header.setQueueId(0);
+ header.setSysFlag(0);
+ header.setBornTimestamp(System.currentTimeMillis());
+ header.setFlag(0);
+ header.setReconsumeTimes(0);
+ header.setUnitMode(false);
+ header.setBatch(false);
+ RemotingCommand send = RemotingCommand.createRequestCommand(RequestCode.SEND_MESSAGE, header);
+ send.setBody(("iotdb-metric-" + i).getBytes(StandardCharsets.UTF_8));
+ assertSuccess(client.invokeSync(send, 3000));
+ }
+ SubscriptionGroupConfig group = new SubscriptionGroupConfig();
+ group.setGroupName(topic);
+ RemotingCommand createGroup = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP, null);
+ createGroup.setBody(RemotingSerializable.encode(group));
+ assertSuccess(client.invokeSync(createGroup, 3000));
+ org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData consumer =
+ new org.apache.rocketmq.remoting.protocol.heartbeat.ConsumerData();
+ consumer.setGroupName(topic);
+ consumer.setConsumeType(org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType.CONSUME_ACTIVELY);
+ consumer.setMessageModel(org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel.CLUSTERING);
+ consumer.setConsumeFromWhere(org.apache.rocketmq.common.consumer.ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
+ consumer.getSubscriptionDataSet().add(org.apache.rocketmq.remoting.protocol.filter.FilterAPI.buildSubscriptionData(topic, "*"));
+ heartbeat.getConsumerDataSet().add(consumer);
+ RemotingCommand consumerBeat = RemotingCommand.createRequestCommand(RequestCode.HEART_BEAT, null);
+ consumerBeat.setVersion(org.apache.rocketmq.common.MQVersion.CURRENT_VERSION);
+ consumerBeat.setBody(RemotingSerializable.encode(heartbeat));
+ assertSuccess(client.invokeSync(consumerBeat, 3000));
+ org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader pull =
+ new org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader();
+ pull.setConsumerGroup(topic);
+ pull.setTopic(topic);
+ pull.setQueueId(0);
+ pull.setQueueOffset(0L);
+ pull.setMaxMsgNums(2);
+ pull.setSysFlag(org.apache.rocketmq.common.sysflag.PullSysFlag.buildSysFlag(false, false, true, false));
+ pull.setCommitOffset(0L);
+ pull.setSuspendTimeoutMillis(0L);
+ pull.setSubscription("*");
+ pull.setSubVersion(System.currentTimeMillis());
+ pull.setExpressionType("TAG");
+ assertSuccess(client.invokeSync(RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, pull), 3000));
+ UpdateConsumerOffsetRequestHeader committed = new UpdateConsumerOffsetRequestHeader();
+ committed.setConsumerGroup(topic);
+ committed.setTopic(topic);
+ committed.setQueueId(0);
+ committed.setCommitOffset(2L);
+ assertSuccess(client.invokeSync(RemotingCommand.createRequestCommand(RequestCode.UPDATE_CONSUMER_OFFSET, committed), 3000));
+ }
+
+ private RuntimeMetadata runtime() {
+ RuntimeMetadata runtime = new RuntimeMetadata();
+ runtime.setId(System.nanoTime());
+ runtime.setClusterId(1L);
+ runtime.setClusterType(ClusterType.STORAGE_ROCKETMQ_BROKER_MAIN_SLAVE);
+ return runtime;
+ }
+
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org