feat(kafka): add consumer group backlog view

crates/dbx-core/src/mq/adapters/kafka.rs

@@ -301,6 +301,10 @@ impl MessageQueueAdmin for KafkaAdmin {
         Ok(subs)
     }
 
+    async fn get_kafka_consumer_group_snapshot(&self) -> Result<KafkaConsumerGroupSnapshot, String> {
+        self.call("mq_get_consumer_group_snapshot", consumer_group_snapshot_params(&self.config)).await
+    }
+
     async fn create_subscription(&self, _topic: &TopicRef, _sub: &str, _pos: ResetPosition) -> Result<(), String> {
         Err("Kafka consumer groups are created automatically when consumers join".to_string())
     }
@@ -657,6 +661,12 @@ fn build_connection_params(cfg: &MqAdminConfig) -> serde_json::Value {
     })
 }
 
+fn consumer_group_snapshot_params(cfg: &MqAdminConfig) -> serde_json::Value {
+    serde_json::json!({
+        "timeout_ms": cfg.request_timeout_ms(),
+    })
+}
+
 fn peek_messages_params(
     cfg: &MqAdminConfig,
     topic: &TopicRef,
@@ -864,6 +874,28 @@ mod tests {
         assert_eq!(params.get("offset").and_then(|value| value.as_i64()), Some(17));
     }
 
+    #[test]
+    fn consumer_group_snapshot_params_forward_the_configured_query_timeout() {
+        let mut cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false);
+        cfg.query_timeout_secs = 17;
+
+        let params = consumer_group_snapshot_params(&cfg);
+
+        assert_eq!(params.get("timeout_ms").and_then(serde_json::Value::as_u64), Some(17_000));
+        assert_eq!(cfg.rpc_timeout(), Some(std::time::Duration::from_secs(17)));
+    }
+
+    #[test]
+    fn consumer_group_snapshot_params_keep_a_finite_agent_budget_for_unlimited_rpc_timeout() {
+        let mut cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false);
+        cfg.query_timeout_secs = 0;
+
+        let params = consumer_group_snapshot_params(&cfg);
+
+        assert_eq!(params.get("timeout_ms").and_then(serde_json::Value::as_u64), Some(3_600_000));
+        assert_eq!(cfg.rpc_timeout(), None);
+    }
+
     #[test]
     fn peek_message_params_omit_legacy_optional_fields() {
         let cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false);

crates/dbx-core/src/mq/port.rs

@@ -156,6 +156,13 @@ pub trait MessageQueueAdmin: Send + Sync {
         self.list_subscriptions(topic).await
     }
 
+    /// Kafka-only cluster-wide consumer group snapshot. Other MQ systems keep
+    /// the default unsupported response instead of leaking Kafka concepts into
+    /// their existing subscription APIs.
+    async fn get_kafka_consumer_group_snapshot(&self) -> Result<KafkaConsumerGroupSnapshot, String> {
+        Err("Kafka consumer group snapshots are not supported by this MQ system".to_string())
+    }
+
     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

@@ -399,6 +399,14 @@ pub async fn mq_enrich_subscriptions_core(
     adapter.enrich_subscriptions(&topic).await
 }
 
+pub async fn mq_get_kafka_consumer_group_snapshot_core(
+    state: &AppState,
+    conn_id: &str,
+) -> Result<KafkaConsumerGroupSnapshot, String> {
+    let adapter = get_adapter(state, conn_id).await?;
+    adapter.get_kafka_consumer_group_snapshot().await
+}
+
 pub async fn mq_create_subscription_core(
     state: &AppState,
     conn_id: &str,

crates/dbx-core/src/mq/types.rs

@@ -465,6 +465,50 @@ pub struct SubscriptionInfo {
     pub backlog_unavailable: Option<bool>,
 }
 
+/// One Kafka topic-partition's committed and log-end offsets for a consumer group.
+/// Optional offsets distinguish unavailable data from a healthy zero lag.
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct KafkaConsumerGroupPartitionLag {
+    pub topic: String,
+    pub partition: i32,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub current_offset: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub end_offset: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub lag: Option<i64>,
+}
+
+/// Cluster-wide Kafka consumer group summary with its partition lag details.
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct KafkaConsumerGroupSummary {
+    pub group_id: String,
+    pub state: String,
+    #[serde(default)]
+    pub simple_group: bool,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub member_count: Option<u32>,
+    #[serde(default)]
+    pub topics: Vec<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub total_lag: Option<i64>,
+    #[serde(default)]
+    pub lag_available: bool,
+    #[serde(default)]
+    pub partitions: Vec<KafkaConsumerGroupPartitionLag>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+}
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct KafkaConsumerGroupSnapshot {
+    #[serde(default)]
+    pub groups: Vec<KafkaConsumerGroupSummary>,
+}
+
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ConsumerInfo {
@@ -521,6 +565,41 @@ mod reset_position_tests {
     }
 }
 
+#[cfg(test)]
+mod kafka_consumer_group_snapshot_tests {
+    use super::KafkaConsumerGroupSnapshot;
+
+    #[test]
+    fn preserves_unavailable_offsets_as_none() {
+        let snapshot: KafkaConsumerGroupSnapshot = serde_json::from_value(serde_json::json!({
+            "groups": [{
+                "groupId": "orders-service",
+                "state": "STABLE",
+                "simpleGroup": false,
+                "memberCount": 1,
+                "topics": ["orders"],
+                "totalLag": null,
+                "lagAvailable": false,
+                "partitions": [{
+                    "topic": "orders",
+                    "partition": 0,
+                    "currentOffset": 8,
+                    "endOffset": null,
+                    "lag": null
+                }],
+                "error": "End offsets unavailable for 1 partition(s)"
+            }]
+        }))
+        .expect("deserialize Kafka consumer group snapshot");
+
+        assert_eq!(snapshot.groups[0].total_lag, None);
+        assert!(!snapshot.groups[0].lag_available);
+        assert_eq!(snapshot.groups[0].partitions[0].current_offset, Some(8));
+        assert_eq!(snapshot.groups[0].partitions[0].end_offset, None);
+        assert_eq!(snapshot.groups[0].partitions[0].lag, None);
+    }
+}
+
 /// How many messages to skip on a subscription.
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "camelCase", rename_all_fields = "camelCase")]