github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 14:52:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(rocketmq): 优化消费组加载/连接并修复积压、FIFO 与 Docker remap

Browse Source 
* feat(rocketmq): 优化 RocketMQ 消费者组管理和性能监控

- 添加并行订阅组配置收集功能，支持超时控制和并发限制
- 实现消费者滞后监控功能，支持按主题过滤的滞后统计
- 重构消费滞后结果构建逻辑，添加 Broker 客户端映射功能
- 优化 Admin 客户端构建，确保请求超时参数正确传递
- 为 RabbitMQ 面板组件添加 MQ 写入保护确认机制
- 统一 MQ 面板样式表引入，清理重复按钮样式定义
- 增强连接对话框中的生产环境保护逻辑，支持 MQ 类型范围
- 更新测试用例以验证新的消费者滞后计算和客户端映射功能
- 为 RocketMQ 订阅面板添加过滤计数功能和测试覆盖

* feat(rocketmq): 优化 RocketMQ 客户端连接和消费者组管理

- 移除硬编码的 5 秒收集超时，改用连接配置的请求超时
- 添加从第一个可用主节点收集消费者组配置的方法以提高速度
- 重构内部方法访问客户端 API 实现和超时配置
- 在主题过滤模式下保留一个主节点转储以恢复 FIFO 类型准确性
- 改进消费延迟计算，失败时不设置总延迟而是保持未定义
- 为消费者运行信息获取添加 Docker 地址映射支持
- 更新命名服务器探测逻辑以更好地处理高可用配置
- 增加消息队列保活机制的查询超时支持
- 在前端界面中显示部分积压加载失败的错误信息
- 添加卸载主题前的操作确认保护
- 改进空结果集的用户界面提示信息

* fix(rocketmq): 解决消费组延迟统计失败时显示为健康零延迟的问题

- 在 examineConsumeStatsRemapped 中添加成功计数和最后错误记录
- 添加 ensureConsumeStatsProbeSucceeded 方法确保没有主节点响应时抛出异常
- 修改 TopicConsumeDetail 接口使 delay 字段可选以表示探针失败
- 更新 UI 显示逻辑在延迟不可用时显示破折号而不是零
- 添加测试用例验证所有代理失败时的异常处理
- 在 Rust 代码中检查 totalLag 是否存在以避免将缺失值视为零
- 添加消息队列适配器的标识检查防止过期的保活连接被错误清理

* fix(rocketmq): 修复RocketMQ消费组详情加载和统计查询问题

- 为examineConsumeStats调用添加超时参数以支持Docker地址映射
- 更新ensureConsumeStatsProbeSucceeded方法以处理部分失败情况下的空合并结果
- 在消费者组对话框中添加序列号防止陈旧详情/配置响应覆盖新数据
- 修复消费延迟显示逻辑，将null延迟值显示为"-"
- 改进错误处理，确保仅在所有代理成功时才接受空偏移表作为有效结果

* fix(rocketmq): 修复消费者组配置收集和消息积压显示问题

- 修改 collectConsumerGroupConfigsForTopicFilter 方法以合并多个主节点的订阅配置
- 优化 Topic-filtered 列表的配置收集逻辑，确保 FIFO 组类型准确性
- 添加 totalLagFailed 标志以正确处理消息积压探测失败情况
- 更新前端组件以在探测失败时显示 "-" 而非 0
- 修复 Docker 环境下多代理地址映射冲突问题
- 添加 RFC1918 私有 IP 验证方法并改进主机可达性判断
- 为积压不可用状态添加数据模型支持

* fix(rocketmq): 修复RocketMQ消费者组配置收集中的FIFO组识别问题

- 重构订阅组配置合并逻辑，确保FIFO组不会被误识别为NORMAL组
- 添加mergeSubscriptionGroupConfigs方法处理重复和冲突的订阅组配置
- 修改collectConsumerGroupConfigsForTopicFilter方法使用新的合并策略
- 添加测试用例验证FIFO组优先级和后续组保持功能
- 修复中间代理返回重复转储时丢失FIFO-only组的问题

* refactor(rocketmq): 优化消费组统计和名称服务器连接逻辑

- 引入 MasterBrokerAddrPlan 类来管理重映射和冲突回退地址
- 将 examineConsumeStatsRemapped 重构为 examineConsumeStatsOnMasters 方法
- 实现对 Docker 环境中重映射地址的更好支持和故障转移处理
- 在连接前预先探测 NameServer 可达性以避免 JVM 启动超时
- 优化消费者组在线成员数显示，未知状态时显示为 '-'
- 添加相关单元测试验证地址解析和冲突处理逻辑

* fix(rocketmq): 修复Docker重映射碰撞和消费者组配置问题

- 修复订阅组配置查找逻辑，支持FIFO合并避免顺序消息被隐藏
- 添加超时参数到订阅组查询方法
- 实现Docker重映射碰撞检测和失败关闭机制
- 添加测试验证碰撞空合并失败关闭逻辑
- 修复RocketMQ跟踪详情对话框加载序列控制问题
- 更新消费延迟标签为积压(backlog)
- 添加连接缺失和只读写拒绝的国际化消息提示
- 修复CSS样式中按钮主题颜色应用问题
- 优化NameServer探测预算计算逻辑

* fix(rocketmq): 修复消费者组类型分类和成员计数显示问题

- 当订阅配置缺失时返回 UNKNOWN 类型而不是默认 NORMAL
- 在删除和更新消费者组时对所有主节点执行操作确保一致性
- 成员计数查询失败时保持字段为空避免显示错误的离线状态
- 添加 UNKNOWN 类型的国际化支持和UI样式适配
- 修改测试用例验证缺失配置时的类型分类逻辑

* style(mq): 移除MQ面板中的主按钮样式覆盖

- 删除了 .mq-btn-primary 和 .btn-primary 的填充颜色样式定义
- 保留了悬停效果样式以维持用户体验一致性
- 更新注释说明MQ Chrome在浅色主题中保持页面表面样式
- 避免与工具栏和表格操作按钮产生视觉冲突

* test(composables): 为 useMqMutationGuard 添加测试并更新实现

- 添加了测试用例验证MQ面板使用的getter连接ID解析
- 将 unref 替换为 toValue 以正确处理getter参数
- 更新类型导入使用 MaybeRefOrGetter 替代 MaybeRef
- 添加注释说明使用 toValue 的原因

* fix(rocketmq): 修复消费者连接检查和订阅组变更的容错机制

- 重构 examineConsumerConnectionInfoRemapped 方法以正确处理无主节点响应的情况
- 添加 requireConsumerConnectionProbeResult 方法确保连接探针结果的准确性
- 修改 mutateSubscriptionGroupOnMasters 方法实现全地址覆盖要求防止部分更新
- 新增 ensureSubscriptionGroupMutationSucceeded 和 shouldFailClosedOnCollisionPartialMutation
  方法增强订阅组变更的错误处理
- 更新 rocketmq_subscription_from_group 函数默认类型为 UNKNOWN 而非 NORMAL
- 添加多个单元测试验证连接探针和订阅组变更的各种异常场景

* test(rocketmq): 为消费者连接探测测试方法添加异常声明

- 在requireConsumerConnectionProbeResultFailsClosedWhenNoMasterAnswers测试方法中添加throws Exception声明
- 确保测试方法能够正确处理可能抛出的异常情况

* fix(i18n): MQ 生产确认 source 本地化并清理 fieldStatus 死键

---------

Co-authored-by: AiLi <aili@aili.com>
Co-authored-by: t8y2 <t8y2@users.noreply.github.com>
Co-authored-by: skyler <1156263951@qq.com>
This commit is contained in:
AiLiAiLit8y2skyler
 authored and GitHub committed 2026-08-06 01:12:31 +08:00
1 parent 5c281b1653
commit 2f9ca313cb
61 files changed
+3400 -1099
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/connection.rs
+102 -5
View File
@@ -1292,6 +1292,12 @@ impl AppState {
if wait_for_drain {
self.wait_for_pool_drain(&pool_key).await;
}
#[cfg(feature = "mq-admin")]
let mq_keepalive_adapter = if matches!(&pool, PoolKind::MessageQueue) {
self.mq_registry.get_cached_adapter(&config.id).await
} else {
None
};
let routing = self.pool_routing_control();
let previous = loop {
let mut connections = self.connections.write().await;
@@ -1309,7 +1315,13 @@ impl AppState {
// candidate never reaches this point, so the existing route keeps its state.
routing.stop_keepalive(&pool_key);
activity.insert(pool_key.clone(), PoolActivity::now());
self.start_keepalive_task(&pool_key, &pool, config);
self.start_keepalive_task(
&pool_key,
&pool,
config,
#[cfg(feature = "mq-admin")]
mq_keepalive_adapter.clone(),
);
break Ok(connections.insert(pool_key.clone(), pool));
};
let previous = match previous {
@@ -1430,9 +1442,21 @@ impl AppState {
self.pool_routing_control().close_pool_with_timeout(pool_key, pool).await;
}
fn start_keepalive_task(&self, pool_key: &str, pool: &PoolKind, config: &ConnectionConfig) {
fn start_keepalive_task(
&self,
pool_key: &str,
pool: &PoolKind,
config: &ConnectionConfig,
#[cfg(feature = "mq-admin")] mq_adapter: Option<std::sync::Arc<dyn crate::mq::port::MessageQueueAdmin>>,
) {
let interval_secs = config.keepalive_interval_secs;
let mut target = keepalive_target_from_pool(pool, config);
#[cfg(feature = "mq-admin")]
if target.is_none() {
if let Some(adapter) = mq_adapter {
target = Some(KeepaliveTarget::MessageQueue(adapter));
}
}
if interval_secs == 0 {
return;
}
@@ -1445,10 +1469,28 @@ impl AppState {
let key = pool_key.to_string();
let interval = Duration::from_secs(interval_secs.max(1));
// MQ keepalive runs a full adapter test_connection (agent RPC); use query timeout
// so slow clusters are not spuriously dropped by the shorter connect timeout.
#[cfg(feature = "mq-admin")]
let timeout = if matches!(target, Some(KeepaliveTarget::MessageQueue(_))) {
match config.effective_query_timeout_secs() {
0 => Duration::from_secs(300), // UI "unlimited" still needs a keepalive bound
secs => Duration::from_secs(secs.max(1)),
}
} else {
Duration::from_secs(config.effective_connect_timeout_secs().max(1))
};
#[cfg(not(feature = "mq-admin"))]
let timeout = Duration::from_secs(config.effective_connect_timeout_secs().max(1));
let routing = self.pool_routing_control();
let connections = self.connections.clone();
let running_queries = self.running_queries.clone();
// MQ pool markers are empty; close_pool_kind is a no-op, so keepalive must
// drop the registry adapter or reconnect would reuse a dead agent.
#[cfg(feature = "mq-admin")]
let mq_registry = self.mq_registry.clone();
#[cfg(feature = "mq-admin")]
let mq_connection_id = config.id.clone();
self.task_supervisor.spawn_replace(format!("keepalive:{pool_key}"), move |shutdown| async move {
loop {
tokio::select! {
@@ -1468,6 +1510,17 @@ impl AppState {
log::warn!("Connection keepalive failed for '{key}': {err}; invalidating pool");
let replace_runtime =
err.recovery_decision().is_some_and(RecoveryDecision::replaces_runtime);
// PoolKind::MessageQueue is a unit marker, so matches_pool cannot
// Arc::ptr_eq. Identity-check the registry adapter before teardown.
#[cfg(feature = "mq-admin")]
if let KeepaliveTarget::MessageQueue(adapter) = target {
if !mq_registry.is_current_adapter(&mq_connection_id, adapter).await {
log::debug!("Skipping stale MQ keepalive result for replaced pool '{key}'");
break;
}
}
#[cfg(feature = "mq-admin")]
let drop_mq = matches!(target, KeepaliveTarget::MessageQueue(_));
if !detach_keepalive_target_if_current(
&routing,
&connections,
@@ -1478,6 +1531,11 @@ impl AppState {
.await
{
log::debug!("Skipping stale keepalive result for replaced pool '{key}'");
} else {
#[cfg(feature = "mq-admin")]
if drop_mq {
mq_registry.drop_connection(&mq_connection_id).await;
}
}
break;
}
@@ -1486,8 +1544,22 @@ impl AppState {
"Connection keepalive timed out for '{key}' after {}s; invalidating pool",
timeout.as_secs()
);
#[cfg(feature = "mq-admin")]
if let KeepaliveTarget::MessageQueue(adapter) = target {
if !mq_registry.is_current_adapter(&mq_connection_id, adapter).await {
log::debug!("Skipping stale MQ keepalive timeout for replaced pool '{key}'");
break;
}
}
#[cfg(feature = "mq-admin")]
let drop_mq = matches!(target, KeepaliveTarget::MessageQueue(_));
if !detach_keepalive_target_if_current(&routing, &connections, &key, target, false).await {
log::debug!("Skipping stale keepalive timeout for replaced pool '{key}'");
} else {
#[cfg(feature = "mq-admin")]
if drop_mq {
mq_registry.drop_connection(&mq_connection_id).await;
}
}
break;
}
@@ -3369,7 +3441,13 @@ impl AppState {
}
};
if let (Some(config), Some(client)) = (config, client) {
self.start_keepalive_task(pool_key, &PoolKind::Agent(client), &config);
self.start_keepalive_task(
pool_key,
&PoolKind::Agent(client),
&config,
#[cfg(feature = "mq-admin")]
None,
);
}
}
@@ -4083,7 +4161,10 @@ enum KeepaliveTarget {
Postgres(deadpool_postgres::Pool),
Rqlite(db::rqlite_driver::RqliteClient),
Turso(db::turso_driver::TursoClient),
MongoDb { client: mongodb::Client, database: Option<String> },
MongoDb {
client: mongodb::Client,
database: Option<String>,
},
ClickHouse(db::clickhouse_driver::ChClient),
SqlServer(Arc<tokio::sync::Mutex<db::sqlserver::SqlServerClient>>),
Elasticsearch(db::elasticsearch_driver::EsClient),
@@ -4093,6 +4174,8 @@ enum KeepaliveTarget {
InfluxDb(db::influxdb_driver::InfluxdbClient),
VictoriaMetrics(db::victoriametrics_driver::VictoriaMetricsClient),
Agent(Arc<db::agent_driver::PooledAgentClient>),
#[cfg(feature = "mq-admin")]
MessageQueue(Arc<dyn crate::mq::port::MessageQueueAdmin>),
}
#[derive(Debug)]
@@ -4130,6 +4213,10 @@ impl KeepaliveTarget {
match (self, pool) {
(Self::Agent(expected), PoolKind::Agent(current)) => Arc::ptr_eq(expected, current),
(Self::SqlServer(expected), PoolKind::SqlServer(current)) => Arc::ptr_eq(expected, current),
#[cfg(feature = "mq-admin")]
(Self::MessageQueue(_), PoolKind::MessageQueue) => true,
#[cfg(feature = "mq-admin")]
(Self::MessageQueue(_), _) | (_, PoolKind::MessageQueue) => false,
(Self::Agent(_), _) | (_, PoolKind::Agent(_)) | (Self::SqlServer(_), _) | (_, PoolKind::SqlServer(_)) => {
false
}
@@ -4245,6 +4332,10 @@ async fn ping_keepalive_target(target: &mut KeepaliveTarget, timeout: Duration)
Err(error) => Err(KeepaliveError::Agent(error)),
}
}
#[cfg(feature = "mq-admin")]
KeepaliveTarget::MessageQueue(adapter) => {
adapter.test_connection().await.map(|_| ()).map_err(KeepaliveError::from)
}
}
}
@@ -6553,7 +6644,13 @@ for line in sys.stdin:
.await
.insert(pool_key.to_string(), super::PoolActivity::idle_for(std::time::Duration::from_secs(10)));
let pool = super::clone_pool_kind(state.connections.read().await.get(pool_key).unwrap());
state.start_keepalive_task(pool_key, &pool, &config);
state.start_keepalive_task(
pool_key,
&pool,
&config,
#[cfg(feature = "mq-admin")]
None,
);
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
crates/dbx-core/src/mq/adapters/kafka.rs
+8 -5
View File
@@ -63,12 +63,13 @@ impl KafkaAdmin {
let mut client = AgentDriverClient::spawn(launch).await?;
// Handshake
let _: serde_json::Value = client.call("handshake", serde_json::json!({})).await?;
let _: serde_json::Value =
client.call_with_timeout("handshake", serde_json::json!({}), cfg.rpc_timeout()).await?;
// Build the connection params from MqAdminConfig
let conn_params = build_connection_params(&cfg);
let connect_params = serde_json::json!({ "connection": conn_params });
let _: serde_json::Value = client.call("connect", connect_params).await?;
let _: serde_json::Value = client.call_with_timeout("connect", connect_params, cfg.rpc_timeout()).await?;
log::info!("Kafka admin connected via agent (bootstrap servers: {})", bootstrap_servers(&cfg));
@@ -82,7 +83,7 @@ impl KafkaAdmin {
params: serde_json::Value,
) -> Result<T, String> {
let mut client = self.client.lock().await;
client.call(method, params).await
client.call_with_timeout(method, params, self.config.rpc_timeout()).await
}
/// The Kafka agent bounds message browsing with its configured request timeout.
@@ -538,8 +539,7 @@ impl MessageQueueAdmin for KafkaAdmin {
)
.await?;
let total_lag = result.get("totalLag").and_then(|v| v.as_i64()).unwrap_or(0);
Ok(BacklogStats { msg_backlog: total_lag, backlog_size: 0 })
Ok(backlog_stats_from_consumer_lag(&result))
}
async fn get_cluster_info(&self) -> Result<ClusterInfo, String> {
@@ -655,6 +655,7 @@ fn build_connection_params(cfg: &MqAdminConfig) -> serde_json::Value {
"skip_verify": cfg.tls_skip_verify,
},
"properties": properties,
"request_timeout_ms": cfg.request_timeout_ms(),
})
}
@@ -802,6 +803,8 @@ mod tests {
token_signing: None,
connect_override: None,
management_connect_override: None,
query_timeout_secs: crate::mq::config::DEFAULT_MQ_QUERY_TIMEOUT_SECS,
connect_timeout_secs: crate::mq::config::DEFAULT_MQ_CONNECT_TIMEOUT_SECS,
extra,
}
}
crates/dbx-core/src/mq/adapters/pulsar.rs
+11 -6
View File
@@ -16,8 +16,6 @@ use crate::mq::port::MessageQueueAdmin;
use crate::mq::types::*;
use crate::mq::util::truncate;
/// How long to wait for a single admin REST call.
const REQUEST_TIMEOUT_SECS: u64 = 30;
/// How long to wait for the initial version-probe during construction.
const PROBE_TIMEOUT_SECS: u64 = 8;
const DETAIL_REQUEST_CONCURRENCY: usize = 8;
@@ -39,7 +37,12 @@ impl PulsarAdmin {
/// the API profile. Probe failure degrades gracefully to the 3.1.x baseline.
pub async fn new(cfg: MqAdminConfig) -> Result<Self, String> {
let base = normalize_base(&cfg.admin_url);
let http = build_http_client(cfg.tls_skip_verify, cfg.connect_override.as_ref(), &cfg.admin_url)?;
// Unlimited query timeout still needs a finite HTTP client timeout; mirror
// request_timeout_ms (1h) so Advanced "0 = unlimited" is not silently clamped to 30s.
let request_timeout =
cfg.rpc_timeout().unwrap_or_else(|| std::time::Duration::from_millis(cfg.request_timeout_ms()));
let http =
build_http_client(cfg.tls_skip_verify, cfg.connect_override.as_ref(), &cfg.admin_url, request_timeout)?;
let auth = cfg.auth.clone();
let token_cache = TokenCache::default();
@@ -807,9 +810,9 @@ fn build_http_client(
tls_skip_verify: bool,
connect_override: Option<&MqConnectOverride>,
admin_url: &str,
request_timeout: std::time::Duration,
) -> Result<reqwest::Client, String> {
let mut builder = crate::db::http_client_builder(std::time::Duration::from_secs(REQUEST_TIMEOUT_SECS))
.timeout(std::time::Duration::from_secs(REQUEST_TIMEOUT_SECS));
let mut builder = crate::db::http_client_builder(request_timeout).timeout(request_timeout);
if tls_skip_verify {
builder = builder.danger_accept_invalid_certs(true);
}
@@ -1074,7 +1077,7 @@ fn backlog_stats_from_topic_stats(profile: &PulsarApiProfile, stats: &TopicStats
None => subs.into_iter().fold(0_i64, |total, sub| total.saturating_add(sub.msg_backlog)),
};
BacklogStats { msg_backlog, backlog_size: stats.backlog_size }
BacklogStats { msg_backlog, backlog_size: stats.backlog_size, partitions: Vec::new() }
}
#[cfg(test)]
@@ -1151,6 +1154,8 @@ mod tests {
token_signing: None,
connect_override: None,
management_connect_override: None,
query_timeout_secs: crate::mq::config::DEFAULT_MQ_QUERY_TIMEOUT_SECS,
connect_timeout_secs: crate::mq::config::DEFAULT_MQ_CONNECT_TIMEOUT_SECS,
extra: serde_json::Value::Null,
})
.await
crates/dbx-core/src/mq/adapters/rabbitmq.rs
+8 -4
View File
@@ -64,12 +64,13 @@ impl RabbitMqAdmin {
let mut client = AgentDriverClient::spawn(launch).await?;
// Handshake
let _: serde_json::Value = client.call("handshake", serde_json::json!({})).await?;
let _: serde_json::Value =
client.call_with_timeout("handshake", serde_json::json!({}), cfg.rpc_timeout()).await?;
// Build the connection params from MqAdminConfig
let conn_params = build_connection_params(&cfg)?;
let connect_params = serde_json::json!({ "connection": conn_params });
let _: serde_json::Value = client.call("connect", connect_params).await?;
let _: serde_json::Value = client.call_with_timeout("connect", connect_params, cfg.rpc_timeout()).await?;
log::info!("RabbitMQ admin connected via agent (addresses: {})", addresses(&cfg));
@@ -83,7 +84,7 @@ impl RabbitMqAdmin {
params: serde_json::Value,
) -> Result<T, String> {
let mut client = self.client.lock().await;
client.call(method, params).await
client.call_with_timeout(method, params, self.config.rpc_timeout()).await
}
/// Send a JSON-RPC call that returns `{ok: true}` on success.
@@ -623,7 +624,7 @@ impl MessageQueueAdmin for RabbitMqAdmin {
let result: serde_json::Value = self.call("mq_get_topic_stats", params).await?;
let total_messages = result.get("totalMessages").and_then(|v| v.as_i64()).unwrap_or(0);
Ok(BacklogStats { msg_backlog: total_messages, backlog_size: total_messages })
Ok(BacklogStats { msg_backlog: total_messages, backlog_size: total_messages, partitions: Vec::new() })
}
async fn get_cluster_info(&self) -> Result<ClusterInfo, String> {
@@ -952,6 +953,7 @@ fn build_connection_params(cfg: &MqAdminConfig) -> Result<serde_json::Value, Str
"virtual_host": virtual_host,
"tls_skip_verify": cfg.tls_skip_verify,
"properties": properties,
"request_timeout_ms": cfg.request_timeout_ms(),
});
if let Some(username) = username.filter(|value| !value.trim().is_empty()) {
params["username"] = serde_json::json!(username);
@@ -1027,6 +1029,8 @@ mod tests {
token_signing: None,
connect_override: None,
management_connect_override: None,
query_timeout_secs: crate::mq::config::DEFAULT_MQ_QUERY_TIMEOUT_SECS,
connect_timeout_secs: crate::mq::config::DEFAULT_MQ_CONNECT_TIMEOUT_SECS,
extra,
}
}
crates/dbx-core/src/mq/adapters/rocketmq.rs
+158 -32
View File
@@ -9,11 +9,15 @@
//! 3. Delegate all `MessageQueueAdmin` trait methods to JSON-RPC calls
use std::collections::HashMap;
use std::net::ToSocketAddrs;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use serde::de::DeserializeOwned;
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use tokio::time::timeout;
use crate::db::agent_driver::{AgentDriverClient, AgentLaunchSpec};
use crate::mq::auth::MqAuth;
@@ -94,16 +98,20 @@ fn cluster_info_from_agent_result(result: &serde_json::Value) -> MqClusterInfo {
impl RocketMqAdmin {
/// Spawn the RocketMQ Java agent, perform handshake, and connect.
///
/// Callers must probe NameServer reachability before invoking this so the
/// connect-timeout wall covers only JVM spawn + handshake + connect.
pub async fn new(cfg: MqAdminConfig, launch: AgentLaunchSpec) -> Result<Self, String> {
let mut client = AgentDriverClient::spawn(launch).await?;
// Handshake
let _: serde_json::Value = client.call("handshake", serde_json::json!({})).await?;
let _: serde_json::Value =
client.call_with_timeout("handshake", serde_json::json!({}), cfg.rpc_timeout()).await?;
// Build the connection params from MqAdminConfig
let conn_params = build_connection_params(&cfg);
let connect_params = serde_json::json!({ "connection": conn_params });
let _: serde_json::Value = client.call("connect", connect_params).await?;
let _: serde_json::Value = client.call_with_timeout("connect", connect_params, cfg.rpc_timeout()).await?;
log::info!("RocketMQ admin connected via agent (namesrv: {})", namesrv_addr(&cfg));
@@ -117,7 +125,7 @@ impl RocketMqAdmin {
params: serde_json::Value,
) -> Result<T, String> {
let mut client = self.client.lock().await;
client.call(method, params).await
client.call_with_timeout(method, params, self.config.rpc_timeout()).await
}
/// Send a JSON-RPC call that returns `{ok: true}` on success.
@@ -355,13 +363,14 @@ impl MessageQueueAdmin for RocketMqAdmin {
async fn list_subscriptions(&self, topic: &TopicRef) -> Result<Vec<SubscriptionInfo>, String> {
if topic.topic.is_empty() {
// Fast path: names/types only. UI enrichs online members/topics in a second pass.
let result: serde_json::Value = self
.call(
"mq_list_consumer_groups",
serde_json::json!({
"limit": 500,
"offset": 0,
"enrich": true,
"enrich": false,
}),
)
.await?;
@@ -369,8 +378,7 @@ impl MessageQueueAdmin for RocketMqAdmin {
return Ok(groups.iter().map(rocketmq_subscription_from_group).collect());
}
// Agent uses queryTopicConsumeByWho (Dashboard queryTopicConsumerInfo); include all
// returned groups even when consumers are offline and offsets are zero.
// Topic path: skip list enrich; batch lag in one agent RPC (no N+1).
let result: serde_json::Value = self
.call(
"mq_list_consumer_groups",
@@ -378,34 +386,30 @@ impl MessageQueueAdmin for RocketMqAdmin {
"topic": topic.topic,
"limit": 200,
"offset": 0,
"enrich": true,
"enrich": false,
"includeLag": true,
}),
)
.await?;
let groups = result.get("groups").and_then(|v| v.as_array()).cloned().unwrap_or_default();
Ok(groups.iter().map(rocketmq_subscription_from_group).collect())
}
let mut subs = Vec::new();
for group in groups {
let group_id = group.get("groupId").and_then(|v| v.as_str()).unwrap_or_default();
if group_id.is_empty() {
continue;
}
let mut sub = rocketmq_subscription_from_group(&group);
if let Ok(lag) = self
.call::<serde_json::Value>(
"mq_get_consumer_lag",
serde_json::json!({
"groupId": group_id,
"topic": topic.topic,
}),
)
.await
{
sub.msg_backlog = lag.get("totalLag").and_then(|v| v.as_i64()).unwrap_or(0);
}
subs.push(sub);
async fn enrich_subscriptions(&self, topic: &TopicRef) -> Result<Vec<SubscriptionInfo>, String> {
// Cluster-wide second pass fills memberCount/topics after the fast list paint.
let mut params = serde_json::json!({
"limit": 500,
"offset": 0,
"enrich": true,
});
if !topic.topic.is_empty() {
params["topic"] = serde_json::json!(topic.topic);
params["limit"] = serde_json::json!(200);
params["includeLag"] = serde_json::json!(true);
}
Ok(subs)
let result: serde_json::Value = self.call("mq_list_consumer_groups", params).await?;
let groups = result.get("groups").and_then(|v| v.as_array()).cloned().unwrap_or_default();
Ok(groups.iter().map(rocketmq_subscription_from_group).collect())
}
async fn create_subscription(&self, _topic: &TopicRef, _sub: &str, _pos: ResetPosition) -> Result<(), String> {
@@ -673,8 +677,12 @@ impl MessageQueueAdmin for RocketMqAdmin {
)
.await?;
let total_lag = result.get("totalLag").and_then(|v| v.as_i64()).unwrap_or(0);
Ok(BacklogStats { msg_backlog: total_lag, backlog_size: 0 })
// Agent omits totalLag on probe failure; never coerce that to healthy zero backlog.
if result.get("totalLag").and_then(|v| v.as_i64()).is_none() {
return Err(format!("RocketMQ consumer lag unavailable for group '{group_id}' on topic '{}'", topic.topic));
}
Ok(backlog_stats_from_consumer_lag(&result))
}
async fn get_cluster_info(&self) -> Result<ClusterInfo, String> {
@@ -831,6 +839,7 @@ fn build_connection_params(cfg: &MqAdminConfig) -> serde_json::Value {
"access_key": access_key,
"secret_key": secret_key,
"tls_skip_verify": cfg.tls_skip_verify,
"request_timeout_ms": cfg.request_timeout_ms(),
})
}
@@ -901,12 +910,14 @@ fn rocketmq_subscription_for_topic(
online_members: None,
consumer_group_type: None,
message_model: None,
backlog_unavailable: None,
})
}
fn rocketmq_subscription_from_group(group: &serde_json::Value) -> SubscriptionInfo {
let group_id = group.get("groupId").and_then(|v| v.as_str()).unwrap_or_default();
let group_type = group.get("groupType").and_then(|v| v.as_str()).unwrap_or("NORMAL").to_string();
// Match agent classify: missing dump → UNKNOWN, not silent NORMAL (FIFO hide risk).
let group_type = group.get("groupType").and_then(|v| v.as_str()).unwrap_or("UNKNOWN").to_string();
let message_model = group.get("messageModel").and_then(|v| v.as_str()).map(String::from);
let online_members = group.get("memberCount").and_then(|v| v.as_u64()).map(|v| v as u32);
let topics = group
@@ -914,10 +925,13 @@ fn rocketmq_subscription_from_group(group: &serde_json::Value) -> SubscriptionIn
.and_then(|v| v.as_array())
.map(|arr| arr.iter().filter_map(|v| v.as_str().map(String::from)).collect::<Vec<_>>())
.unwrap_or_default();
let lag_failed = group.get("totalLagFailed").and_then(|v| v.as_bool()).unwrap_or(false);
let total_lag = group.get("totalLag").and_then(|v| v.as_i64());
SubscriptionInfo {
name: group_id.to_string(),
sub_type: group_type.clone(),
msg_backlog: 0,
// Probe failure: keep 0 but set backlog_unavailable so topic-list UI shows "-".
msg_backlog: total_lag.unwrap_or(0),
msg_rate_out: 0.0,
msg_throughput_out: 0.0,
consumers: Vec::new(),
@@ -925,9 +939,92 @@ fn rocketmq_subscription_from_group(group: &serde_json::Value) -> SubscriptionIn
online_members,
consumer_group_type: Some(group_type),
message_model,
backlog_unavailable: if lag_failed || (group.get("totalLag").is_some() && total_lag.is_none()) {
Some(true)
} else {
None
},
}
}
/// Fail fast on unreachable NameServer before paying for JVM agent startup.
/// Called outside the connect-timeout wall so the probe does not steal JVM budget.
pub(crate) async fn probe_namesrv_before_connect(cfg: &MqAdminConfig, budget: Duration) -> Result<(), String> {
probe_namesrv_tcp(&namesrv_addr(cfg), budget).await
}
/// Probe NameServer addresses so connect fails before spawning the JVM agent.
/// Tries each `;`-separated host (and each resolved IP) within the shared budget,
/// matching RocketMQ client HA behavior instead of failing on the first dead node.
async fn probe_namesrv_tcp(namesrv: &str, budget: Duration) -> Result<(), String> {
let hosts: Vec<String> =
namesrv.split(';').map(str::trim).filter(|part| !part.is_empty()).map(str::to_string).collect();
if hosts.is_empty() {
return Err("RocketMQ namesrv_addr is empty".to_string());
}
let deadline = tokio::time::Instant::now() + budget;
// Multi-NS HA: cap each attempt so a blackholed first node leaves budget for later hosts.
let per_attempt = if hosts.len() <= 1 {
budget
} else {
(budget / hosts.len() as u32).clamp(Duration::from_millis(500), Duration::from_secs(3))
};
let mut last_error = String::new();
for host in &hosts {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
break;
}
// DNS resolution is blocking; keep it off the async runtime.
let host_for_resolve = host.clone();
let addrs = match tokio::task::spawn_blocking(move || {
host_for_resolve
.to_socket_addrs()
.map(|iter| iter.collect::<Vec<_>>())
.map_err(|e| format!("RocketMQ NameServer address '{host_for_resolve}' is invalid: {e}"))
})
.await
{
Ok(Ok(addrs)) => addrs,
Ok(Err(e)) => {
last_error = e;
continue;
}
Err(e) => {
last_error = format!("RocketMQ NameServer resolve task failed: {e}");
continue;
}
};
if addrs.is_empty() {
last_error = format!("RocketMQ NameServer address '{host}' did not resolve");
continue;
}
for addr in addrs {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
break;
}
let attempt = remaining.min(per_attempt);
match timeout(attempt, TcpStream::connect(addr)).await {
Ok(Ok(_stream)) => return Ok(()),
Ok(Err(e)) => {
last_error = format!("Cannot reach RocketMQ NameServer {host}: {e}");
}
Err(_) => {
last_error =
format!("RocketMQ NameServer {host} connect timed out after {}ms", attempt.as_millis());
}
}
}
}
Err(if last_error.is_empty() {
format!("RocketMQ NameServer connect timed out after {}s", budget.as_secs())
} else {
last_error
})
}
fn peeked_message_from_agent_json(idx: usize, message: &serde_json::Value) -> PeekedMessage {
let mut properties = HashMap::new();
if let Some(partition) = message.get("partition").and_then(|v| v.as_i64()) {
@@ -975,10 +1072,18 @@ mod tests {
token_signing: None,
connect_override: None,
management_connect_override: None,
query_timeout_secs: crate::mq::config::DEFAULT_MQ_QUERY_TIMEOUT_SECS,
connect_timeout_secs: crate::mq::config::DEFAULT_MQ_CONNECT_TIMEOUT_SECS,
extra,
}
}
#[tokio::test]
async fn probe_namesrv_rejects_empty_address_list() {
let err = probe_namesrv_tcp(" ; ; ", Duration::from_millis(200)).await.expect_err("empty");
assert!(err.contains("empty"), "{err}");
}
#[test]
fn connection_params_map_namesrv_and_acl_credentials() {
let cfg = rocketmq_config(
@@ -995,6 +1100,7 @@ mod tests {
assert_eq!(params.get("cluster_name").and_then(|v| v.as_str()), Some("DefaultCluster"));
assert_eq!(params.get("access_key").and_then(|v| v.as_str()), Some("rocket"));
assert_eq!(params.get("secret_key").and_then(|v| v.as_str()), Some("secret"));
assert_eq!(params.get("request_timeout_ms").and_then(|v| v.as_u64()), Some(30_000));
}
#[test]
@@ -1031,6 +1137,26 @@ mod tests {
assert_eq!(sub.msg_backlog, 7);
}
#[test]
fn rocketmq_subscription_from_group_marks_failed_lag_unavailable() {
let group = serde_json::json!({
"groupId": "lag-fail",
"groupType": "NORMAL",
"totalLagFailed": true
});
let sub = rocketmq_subscription_from_group(&group);
assert_eq!(sub.msg_backlog, 0);
assert_eq!(sub.backlog_unavailable, Some(true));
}
#[test]
fn rocketmq_subscription_from_group_defaults_missing_type_to_unknown() {
let group = serde_json::json!({ "groupId": "no-type" });
let sub = rocketmq_subscription_from_group(&group);
assert_eq!(sub.sub_type, "UNKNOWN");
assert_eq!(sub.consumer_group_type.as_deref(), Some("UNKNOWN"));
}
#[test]
fn rocketmq_subscription_from_group_maps_offline_topic_consumer() {
let group = serde_json::json!({
crates/dbx-core/src/mq/config.rs
+58
View File
@@ -4,12 +4,19 @@
//! `ConnectionConfig` rather than adding top-level fields, keeping the 50+
//! database-type connection model untouched.
use std::time::Duration;
use serde::{Deserialize, Serialize};
use crate::models::connection::ConnectionConfig;
use crate::mq::auth::MqAuth;
use crate::mq::types::{MqSystemKind, MqTokenSigningConfig};
/// Default query timeout when constructing test configs without a ConnectionConfig.
pub const DEFAULT_MQ_QUERY_TIMEOUT_SECS: u64 = 30;
/// Default connect timeout when constructing test configs without a ConnectionConfig.
pub const DEFAULT_MQ_CONNECT_TIMEOUT_SECS: u64 = 10;
/// Runtime TCP endpoint override for an MQ transport.
///
/// The logical broker endpoint remains unchanged so TLS hostname verification,
@@ -53,6 +60,12 @@ pub struct MqAdminConfig {
/// its Management HTTP API listens on an independently configured port.
#[serde(skip)]
pub management_connect_override: Option<MqConnectOverride>,
/// Runtime-only: from `ConnectionConfig.query_timeout_secs` (`0` = unlimited).
#[serde(skip)]
pub query_timeout_secs: u64,
/// Runtime-only: from `ConnectionConfig.effective_connect_timeout_secs()`.
#[serde(skip)]
pub connect_timeout_secs: u64,
/// System-specific extension fields (e.g. Kafka bootstrap servers).
#[serde(default, skip_serializing_if = "serde_json::Value::is_null")]
pub extra: serde_json::Value,
@@ -76,9 +89,35 @@ impl MqAdminConfig {
{
return Err("Message queue admin URL is empty".to_string());
}
// Advanced connection timeouts live on ConnectionConfig, not external_config.
parsed.query_timeout_secs = cfg.effective_query_timeout_secs();
parsed.connect_timeout_secs = cfg.effective_connect_timeout_secs();
Ok(parsed)
}
/// Agent / HTTP RPC wall-clock timeout. `None` disables the client-side timeout.
pub fn rpc_timeout(&self) -> Option<Duration> {
if self.query_timeout_secs == 0 {
None
} else {
Some(Duration::from_secs(self.query_timeout_secs.max(1)))
}
}
/// Milliseconds for agent `request_timeout_ms`. Unlimited query timeout maps to a
/// large but finite admin request budget so Java/Go clients still make progress.
pub fn request_timeout_ms(&self) -> u64 {
match self.query_timeout_secs {
0 => 3_600_000,
secs => secs.saturating_mul(1000).max(1_000),
}
}
/// Budget for establishing the MQ adapter (agent spawn + handshake/connect).
pub fn connect_timeout(&self) -> Duration {
Duration::from_secs(self.connect_timeout_secs.max(1))
}
pub fn token_signing_configured(&self) -> bool {
self.token_signing.as_ref().is_some_and(MqTokenSigningConfig::is_configured)
}
@@ -228,6 +267,25 @@ mod tests {
assert_eq!(mqc.system_kind, MqSystemKind::RocketMq);
assert_eq!(mqc.admin_url, "");
assert_eq!(mqc.extra.get("namesrvAddr").and_then(|v| v.as_str()), Some("127.0.0.1:9876"));
assert_eq!(mqc.query_timeout_secs, 30);
assert_eq!(mqc.connect_timeout_secs, 5);
assert_eq!(mqc.request_timeout_ms(), 30_000);
}
#[test]
fn copies_advanced_timeouts_from_connection_config() {
let mut cfg = connection_with_external(serde_json::json!({
"systemKind": "rocketmq",
"adminUrl": "",
"extra": { "namesrvAddr": "127.0.0.1:9876" }
}));
cfg.query_timeout_secs = 120;
cfg.connect_timeout_secs = 15;
let mqc = MqAdminConfig::from_connection(&cfg).expect("parse");
assert_eq!(mqc.query_timeout_secs, 120);
assert_eq!(mqc.connect_timeout_secs, 15);
assert_eq!(mqc.request_timeout_ms(), 120_000);
assert_eq!(mqc.rpc_timeout(), Some(std::time::Duration::from_secs(120)));
}
#[test]
crates/dbx-core/src/mq/mod.rs
+37 -6
View File
@@ -45,10 +45,11 @@ pub use crate::mq::types::*;
/// Each connection has its own build-lock so concurrent first-use requests for
/// the same connection block until the first builder finishes, rather than both
/// racing to construct an adapter.
#[derive(Default)]
/// Clone shares the same cache (needed for keepalive tasks that must drop adapters).
#[derive(Clone, Default)]
pub struct MqAdminRegistry {
instances: RwLock<HashMap<String, CachedMqAdmin>>,
build_locks: RwLock<HashMap<String, Arc<Mutex<()>>>>,
instances: Arc<RwLock<HashMap<String, CachedMqAdmin>>>,
build_locks: Arc<RwLock<HashMap<String, Arc<Mutex<()>>>>>,
}
struct CachedMqAdmin {
@@ -65,7 +66,7 @@ pub struct MqBuildResult {
impl MqAdminRegistry {
pub fn new() -> Self {
Self { instances: RwLock::new(HashMap::new()), build_locks: RwLock::new(HashMap::new()) }
Self { instances: Arc::new(RwLock::new(HashMap::new())), build_locks: Arc::new(RwLock::new(HashMap::new())) }
}
/// Return the cached adapter for this connection, building it from the
@@ -108,7 +109,7 @@ impl MqAdminRegistry {
// Config changed — drop the stale adapter so its agent process is released.
self.instances.write().await.remove(connection_id);
let adapter = build_adapter(mqc, agent_launch).await?;
let adapter = build_adapter_with_connect_timeout(mqc, agent_launch).await?;
self.instances
.write()
.await
@@ -116,6 +117,17 @@ impl MqAdminRegistry {
Ok(MqBuildResult { adapter, was_cached: false })
}
/// Cached adapter for keepalive / diagnostics. Returns `None` when not built yet.
pub async fn get_cached_adapter(&self, connection_id: &str) -> Option<Arc<dyn MessageQueueAdmin>> {
self.instances.read().await.get(connection_id).map(|entry| entry.adapter.clone())
}
/// Whether `adapter` is still the live registry entry for this connection (Arc identity).
/// Used so a stale keepalive cannot drop a replacement built after reconnect.
pub async fn is_current_adapter(&self, connection_id: &str, adapter: &Arc<dyn MessageQueueAdmin>) -> bool {
self.instances.read().await.get(connection_id).is_some_and(|entry| Arc::ptr_eq(&entry.adapter, adapter))
}
/// Drop the cached adapter for a connection (called on disconnect).
pub async fn drop_connection(&self, connection_id: &str) {
self.instances.write().await.remove(connection_id);
@@ -144,7 +156,26 @@ impl MqAdminRegistry {
mqc: MqAdminConfig,
agent_launch: Option<AgentLaunchSpec>,
) -> Result<Arc<dyn MessageQueueAdmin>, String> {
build_adapter(mqc, agent_launch).await
build_adapter_with_connect_timeout(mqc, agent_launch).await
}
}
async fn build_adapter_with_connect_timeout(
mqc: MqAdminConfig,
agent_launch: Option<AgentLaunchSpec>,
) -> Result<Arc<dyn MessageQueueAdmin>, String> {
let budget = mqc.connect_timeout();
// RocketMQ: TCP-probe NameServer outside the connect wall so cold JVM spawn
// retains the full connect_timeout (probe used to steal up to half of it).
if mqc.system_kind == MqSystemKindInternal::RocketMq {
// Probe runs outside the connect wall; scale with connect_timeout so HA NameServer
// lists are not starved by a hard 2s cap, but keep an upper bound for snappy UX.
let probe_budget = (budget / 2).clamp(std::time::Duration::from_millis(500), std::time::Duration::from_secs(5));
crate::mq::adapters::rocketmq::probe_namesrv_before_connect(&mqc, probe_budget).await?;
}
match tokio::time::timeout(budget, build_adapter(mqc, agent_launch)).await {
Ok(result) => result,
Err(_) => Err(format!("Message queue connect timed out after {}s", budget.as_secs())),
}
}
crates/dbx-core/src/mq/port.rs
+7
View File
@@ -149,6 +149,13 @@ pub trait MessageQueueAdmin: Send + Sync {
// ---- Subscriptions ----
async fn list_subscriptions(&self, topic: &TopicRef) -> Result<Vec<SubscriptionInfo>, String>;
/// Second-pass enrichment for subscription rows (e.g. RocketMQ online members/topics).
/// Default reuses [`list_subscriptions`]; RocketMQ overrides with `enrich: true`.
async fn enrich_subscriptions(&self, topic: &TopicRef) -> Result<Vec<SubscriptionInfo>, String> {
self.list_subscriptions(topic).await
}
async fn create_subscription(&self, topic: &TopicRef, sub: &str, pos: ResetPosition) -> Result<(), String>;
async fn delete_subscription(&self, topic: &TopicRef, sub: &str, force: bool) -> Result<(), String>;
async fn skip_messages(&self, topic: &TopicRef, sub: &str, count: SkipCount) -> Result<(), String>;
crates/dbx-core/src/mq/service.rs
+34
View File
@@ -390,6 +390,15 @@ pub async fn mq_list_subscriptions_core(
adapter.list_subscriptions(&topic).await
}
pub async fn mq_enrich_subscriptions_core(
state: &AppState,
conn_id: &str,
topic: TopicRef,
) -> Result<Vec<SubscriptionInfo>, String> {
let adapter = get_adapter(state, conn_id).await?;
adapter.enrich_subscriptions(&topic).await
}
pub async fn mq_create_subscription_core(
state: &AppState,
conn_id: &str,
@@ -842,6 +851,8 @@ async fn ensure_connection_writable(state: &AppState, conn_id: &str, operation:
config.name, operation
));
}
// Production protection for desktop MQ uses UI confirmation; MCP enforces
// is_production separately. Do not hard-block confirmed desktop writes here.
}
Ok(())
}
@@ -939,6 +950,29 @@ mod tests {
(state, dir)
}
#[tokio::test]
async fn mutating_rocketmq_send_blocks_read_only_connections() {
let (state, _dir) = test_state_with(mq_connection(true)).await;
let err = mq_send_message_core(
&state,
"readonly-mq",
SendMessageRequest {
topic: "t".into(),
key: None,
payload_base64: "eA==".into(),
payload_text: Some("x".into()),
headers: Default::default(),
partition: None,
exchange: None,
routing_key: None,
namespace: None,
},
)
.await
.expect_err("read-only must block send");
assert!(err.contains("Read-only mode"), "{err}");
}
#[tokio::test]
async fn mutating_service_calls_block_read_only_connections_before_adapter_build() {
let (state, dir) = test_state_with(mq_connection(true)).await;
crates/dbx-core/src/mq/types.rs
+116
View File
@@ -415,6 +415,9 @@ pub struct SubscriptionInfo {
/// RocketMQ: CLUSTERING / BROADCASTING.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub message_model: Option<String>,
/// When true, backlog probe failed — UI must not treat `msg_backlog` as healthy zero.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub backlog_unavailable: Option<bool>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -465,11 +468,76 @@ pub enum SkipCount {
Count { count: u32 },
}
/// Per-queue/partition consume progress (RocketMQ Dashboard consume-detail / Kafka lag rows).
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PartitionBacklog {
pub partition: i32,
/// Consumer committed offset (`consumerOffset` / `currentOffset`).
pub current_offset: i64,
/// Broker max offset (`brokerOffset` / `endOffset`).
pub end_offset: i64,
pub lag: i64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub broker_name: Option<String>,
/// Last consume message store timestamp (ms). `0` means unavailable.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub last_timestamp: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub consumer_client: Option<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct BacklogStats {
pub msg_backlog: i64,
pub backlog_size: i64,
/// Optional queue-level progress; empty for adapters that only expose totals.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub partitions: Vec<PartitionBacklog>,
}
/// Parse agent `mq_get_consumer_lag` JSON (`totalLag` + `partitions[]`) into [`BacklogStats`].
pub fn backlog_stats_from_consumer_lag(value: &serde_json::Value) -> BacklogStats {
let msg_backlog = value.get("totalLag").and_then(|v| v.as_i64()).unwrap_or(0);
let partitions = value
.get("partitions")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|row| {
let partition = row.get("partition").and_then(|v| v.as_i64()).map(|v| v as i32)?;
let current_offset = row.get("currentOffset").and_then(|v| v.as_i64()).unwrap_or(0);
let end_offset = row.get("endOffset").and_then(|v| v.as_i64()).unwrap_or(0);
let lag =
row.get("lag").and_then(|v| v.as_i64()).unwrap_or_else(|| (end_offset - current_offset).max(0));
let broker_name = row
.get("brokerName")
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string);
let last_timestamp = row.get("lastTimestamp").and_then(|v| v.as_i64());
let consumer_client = row
.get("consumerClient")
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string);
Some(PartitionBacklog {
partition,
current_offset,
end_offset,
lag,
broker_name,
last_timestamp,
consumer_client,
})
})
.collect()
})
.unwrap_or_default();
BacklogStats { msg_backlog, backlog_size: 0, partitions }
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -1250,4 +1318,52 @@ mod tests {
assert!(!legacy.incomplete);
assert!(legacy.messages.is_empty());
}
#[test]
fn backlog_stats_from_consumer_lag_maps_dashboard_fields() {
let lag = serde_json::json!({
"totalLag": 10,
"partitions": [
{
"partition": 0,
"currentOffset": 90,
"endOffset": 100,
"lag": 10,
"brokerName": "broker-a",
"lastTimestamp": 1725000000000_i64,
"consumerClient": "172.18.2.212@7#1"
},
{
"partition": 1,
"currentOffset": 50,
"endOffset": 50,
"lag": 0,
"brokerName": "",
"lastTimestamp": 0,
"consumerClient": ""
}
]
});
let stats = super::backlog_stats_from_consumer_lag(&lag);
assert_eq!(stats.msg_backlog, 10);
assert_eq!(stats.partitions.len(), 2);
assert_eq!(stats.partitions[0].broker_name.as_deref(), Some("broker-a"));
assert_eq!(stats.partitions[0].consumer_client.as_deref(), Some("172.18.2.212@7#1"));
assert_eq!(stats.partitions[0].last_timestamp, Some(1725000000000));
// Empty strings are normalized to None so UI can show "-".
assert!(stats.partitions[1].broker_name.is_none());
assert!(stats.partitions[1].consumer_client.is_none());
assert_eq!(stats.partitions[1].last_timestamp, Some(0));
let json = serde_json::to_value(&stats).expect("serialize backlog stats");
assert_eq!(json.get("msgBacklog").and_then(|v| v.as_i64()), Some(10));
assert!(json.get("partitions").and_then(|v| v.as_array()).is_some());
// Legacy callers that only had totals still deserialize with empty partitions.
let legacy: super::BacklogStats =
serde_json::from_value(serde_json::json!({ "msgBacklog": 3, "backlogSize": 0 }))
.expect("legacy backlog stats");
assert_eq!(legacy.msg_backlog, 3);
assert!(legacy.partitions.is_empty());
}
}