feat(kafka): redesign message browsing

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

@@ -85,6 +85,17 @@ impl KafkaAdmin {
         client.call(method, params).await
     }
 
+    /// The Kafka agent bounds message browsing with its configured request timeout.
+    /// Do not preempt that with the driver's generic 30-second RPC timeout.
+    async fn call_with_agent_timeout<T: DeserializeOwned + Send + 'static>(
+        &self,
+        method: &str,
+        params: serde_json::Value,
+    ) -> Result<T, String> {
+        let mut client = self.client.lock().await;
+        client.call_with_timeout(method, params, None).await
+    }
+
     /// Send a JSON-RPC call that returns `{ok: true}` on success.
     async fn call_ok(&self, method: &str, params: serde_json::Value) -> Result<(), String> {
         let _: serde_json::Value = self.call(method, params).await?;
@@ -317,51 +328,11 @@ impl MessageQueueAdmin for KafkaAdmin {
         _sub: &str,
         count: u32,
         options: PeekMessagesOptions,
-    ) -> Result<Vec<PeekedMessage>, String> {
-        let conn_params = build_connection_params(&self.config);
-        let mut params = serde_json::json!({
-            "topic": topic.topic,
-            "count": count,
-            "connection": conn_params,
-        });
-        // Omit partition/offset so the agent defaults to all partitions + earliest.
-        // Do not coerce missing values to 0 — that forced PARTITION 0 OFFSET 0 UX.
-        if let Some(partition) = options.partition {
-            params["partition"] = serde_json::json!(partition);
-        }
-        if let Some(offset) = options.offset {
-            params["offset"] = serde_json::json!(offset);
-        }
-        let result: serde_json::Value = self.call("mq_peek_messages", params).await?;
+    ) -> Result<PeekMessagesResult, String> {
+        let params = peek_messages_params(&self.config, topic, count, options);
+        let result: serde_json::Value = self.call_with_agent_timeout("mq_peek_messages", params).await?;
 
-        let messages = result.get("messages").and_then(|v| v.as_array()).cloned().unwrap_or_default();
-        Ok(messages
-            .into_iter()
-            .enumerate()
-            .map(|(idx, m)| {
-                let mut properties = HashMap::new();
-                if let Some(partition) = m.get("partition").and_then(|v| v.as_i64()) {
-                    properties.insert("partition".to_string(), partition.to_string());
-                }
-                PeekedMessage {
-                    position: (idx + 1) as u32,
-                    message_id: m.get("offset").and_then(|v| v.as_i64()).map(|v| v.to_string()),
-                    key: m.get("key").and_then(|v| v.as_str()).map(String::from),
-                    publish_time: m.get("timestamp").and_then(|v| v.as_i64()).map(|v| v.to_string()),
-                    event_time: None,
-                    properties,
-                    headers: m
-                        .get("headers")
-                        .and_then(|v| v.as_object())
-                        .map(|obj| {
-                            obj.iter().map(|(k, v)| (k.clone(), v.as_str().unwrap_or_default().to_string())).collect()
-                        })
-                        .unwrap_or_default(),
-                    payload_base64: m.get("payloadBase64").and_then(|v| v.as_str()).unwrap_or_default().to_string(),
-                    payload_text: m.get("payloadText").and_then(|v| v.as_str()).map(String::from),
-                }
-            })
-            .collect())
+        Ok(peek_messages_result_from_agent(&result))
     }
 
     async fn expire_messages(&self, _topic: &TopicRef, _sub: &str, _expire_seconds: i64) -> Result<(), String> {
@@ -687,6 +658,70 @@ fn build_connection_params(cfg: &MqAdminConfig) -> serde_json::Value {
     })
 }
 
+fn peek_messages_params(
+    cfg: &MqAdminConfig,
+    topic: &TopicRef,
+    count: u32,
+    options: PeekMessagesOptions,
+) -> serde_json::Value {
+    let mut params = serde_json::json!({
+        "topic": topic.topic,
+        "count": count,
+        "connection": build_connection_params(cfg),
+    });
+    // Keep absent fields absent so older earliest and offset requests retain
+    // their established agent semantics, including offset 0.
+    if let Some(start_position) = options.start_position {
+        params["startPosition"] = serde_json::json!(start_position);
+    }
+    if let Some(partition) = options.partition {
+        params["partition"] = serde_json::json!(partition);
+    }
+    if let Some(offset) = options.offset {
+        params["offset"] = serde_json::json!(offset);
+    }
+    params
+}
+
+fn peek_messages_result_from_agent(result: &serde_json::Value) -> PeekMessagesResult {
+    let (messages, incomplete) = if let Some(messages) = result.as_array() {
+        (messages.clone(), false)
+    } else {
+        (
+            result.get("messages").and_then(|v| v.as_array()).cloned().unwrap_or_default(),
+            result.get("incomplete").and_then(|v| v.as_bool()).unwrap_or(false),
+        )
+    };
+    let messages = messages
+        .into_iter()
+        .enumerate()
+        .map(|(idx, m)| {
+            let mut properties = HashMap::new();
+            if let Some(partition) = m.get("partition").and_then(|v| v.as_i64()) {
+                properties.insert("partition".to_string(), partition.to_string());
+            }
+            PeekedMessage {
+                position: (idx + 1) as u32,
+                message_id: m.get("offset").and_then(|v| v.as_i64()).map(|v| v.to_string()),
+                key: m.get("key").and_then(|v| v.as_str()).map(String::from),
+                publish_time: m.get("timestamp").and_then(|v| v.as_i64()).map(|v| v.to_string()),
+                event_time: None,
+                properties,
+                headers: m
+                    .get("headers")
+                    .and_then(|v| v.as_object())
+                    .map(|obj| {
+                        obj.iter().map(|(k, v)| (k.clone(), v.as_str().unwrap_or_default().to_string())).collect()
+                    })
+                    .unwrap_or_default(),
+                payload_base64: m.get("payloadBase64").and_then(|v| v.as_str()).unwrap_or_default().to_string(),
+                payload_text: m.get("payloadText").and_then(|v| v.as_str()).map(String::from),
+            }
+        })
+        .collect();
+    PeekMessagesResult { messages, incomplete }
+}
+
 fn reset_cursor_params(topic: &TopicRef, sub: &str, pos: ResetPosition) -> Result<serde_json::Value, String> {
     match pos {
         ResetPosition::Earliest => Ok(serde_json::json!({
@@ -771,6 +806,95 @@ mod tests {
         }
     }
 
+    fn topic_ref() -> TopicRef {
+        TopicRef {
+            tenant: "_kafka".to_string(),
+            namespace: "_kafka".to_string(),
+            topic: "events".to_string(),
+            persistent: true,
+            partitioned: None,
+            message_type: None,
+            ..TopicRef::default()
+        }
+    }
+
+    #[test]
+    fn peek_message_params_forward_explicit_start_position_and_offset_filters() {
+        let cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false);
+        let params = peek_messages_params(
+            &cfg,
+            &topic_ref(),
+            20,
+            PeekMessagesOptions {
+                start_position: Some(PeekStartPosition::Offset),
+                partition: Some(2),
+                offset: Some(17),
+            },
+        );
+
+        assert_eq!(params.get("topic").and_then(|value| value.as_str()), Some("events"));
+        assert_eq!(params.get("count").and_then(|value| value.as_u64()), Some(20));
+        assert_eq!(params.get("startPosition").and_then(|value| value.as_str()), Some("offset"));
+        assert_eq!(params.get("partition").and_then(|value| value.as_i64()), Some(2));
+        assert_eq!(params.get("offset").and_then(|value| value.as_i64()), Some(17));
+    }
+
+    #[test]
+    fn peek_message_params_omit_legacy_optional_fields() {
+        let cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false);
+        let params = peek_messages_params(&cfg, &topic_ref(), 20, PeekMessagesOptions::default());
+
+        assert!(params.get("startPosition").is_none());
+        assert!(params.get("partition").is_none());
+        assert!(params.get("offset").is_none());
+    }
+
+    #[test]
+    fn peek_result_preserves_the_agent_incomplete_status() {
+        let result = peek_messages_result_from_agent(&serde_json::json!({
+            "messages": [{
+                "partition": 2,
+                "offset": 17,
+                "payloadBase64": "aGVsbG8=",
+            }],
+            "incomplete": true,
+        }));
+
+        assert!(result.incomplete);
+        assert_eq!(result.messages.len(), 1);
+        assert_eq!(result.messages[0].properties.get("partition").map(String::as_str), Some("2"));
+        assert_eq!(result.messages[0].message_id.as_deref(), Some("17"));
+    }
+
+    #[test]
+    fn peek_result_accepts_legacy_agent_array_responses() {
+        let result = peek_messages_result_from_agent(&serde_json::json!([{
+            "partition": 1,
+            "offset": 9,
+            "payloadBase64": "bGVnYWN5",
+        }]));
+
+        assert!(!result.incomplete);
+        assert_eq!(result.messages.len(), 1);
+        assert_eq!(result.messages[0].properties.get("partition").map(String::as_str), Some("1"));
+        assert_eq!(result.messages[0].message_id.as_deref(), Some("9"));
+        assert_eq!(result.messages[0].payload_base64, "bGVnYWN5");
+    }
+
+    #[test]
+    fn peek_message_params_preserve_legacy_offset_requests() {
+        let cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false);
+        let params = peek_messages_params(
+            &cfg,
+            &topic_ref(),
+            20,
+            PeekMessagesOptions { start_position: None, partition: None, offset: Some(17) },
+        );
+
+        assert!(params.get("startPosition").is_none());
+        assert_eq!(params.get("offset").and_then(|value| value.as_i64()), Some(17));
+    }
+
     #[test]
     fn connection_params_map_basic_auth_to_kafka_sasl_without_plaintext_password_extra() {
         let cfg = kafka_config(

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

@@ -451,9 +451,9 @@ impl MessageQueueAdmin for PulsarAdmin {
         sub: &str,
         count: u32,
         _options: PeekMessagesOptions,
-    ) -> Result<Vec<PeekedMessage>, String> {
+    ) -> Result<PeekMessagesResult, String> {
         if count == 0 {
-            return Ok(Vec::new());
+            return Ok(PeekMessagesResult::default());
         }
 
         let messages: Vec<PeekedMessage> = stream::iter(1..=count)
@@ -477,7 +477,7 @@ impl MessageQueueAdmin for PulsarAdmin {
             .flatten()
             .collect();
 
-        Ok(messages)
+        Ok(PeekMessagesResult::complete(messages))
     }
 
     async fn expire_messages(&self, topic: &TopicRef, sub: &str, expire_seconds: i64) -> Result<(), String> {

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

@@ -517,7 +517,7 @@ impl MessageQueueAdmin for RabbitMqAdmin {
         _sub: &str,
         count: u32,
         _options: PeekMessagesOptions,
-    ) -> Result<Vec<PeekedMessage>, String> {
+    ) -> Result<PeekMessagesResult, String> {
         let conn_params = build_connection_params(&self.config)?;
         require_specific_vhost(&topic.namespace)?;
         let params = with_virtual_host(
@@ -531,7 +531,9 @@ impl MessageQueueAdmin for RabbitMqAdmin {
         let result: serde_json::Value = self.call("mq_peek_messages", params).await?;
 
         let messages = result.get("messages").and_then(|v| v.as_array()).cloned().unwrap_or_default();
-        Ok(messages.into_iter().enumerate().map(|(idx, m)| peeked_message_from_json(idx, &m)).collect())
+        Ok(PeekMessagesResult::complete(
+            messages.into_iter().enumerate().map(|(idx, m)| peeked_message_from_json(idx, &m)).collect(),
+        ))
     }
 
     async fn expire_messages(&self, _topic: &TopicRef, _sub: &str, _expire_seconds: i64) -> Result<(), String> {

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

@@ -444,7 +444,7 @@ impl MessageQueueAdmin for RocketMqAdmin {
         _sub: &str,
         count: u32,
         options: PeekMessagesOptions,
-    ) -> Result<Vec<PeekedMessage>, String> {
+    ) -> Result<PeekMessagesResult, String> {
         let conn_params = build_connection_params(&self.config);
         let mut params = serde_json::json!({
             "topic": topic.topic,
@@ -462,7 +462,9 @@ impl MessageQueueAdmin for RocketMqAdmin {
         let result: serde_json::Value = self.call("mq_peek_messages", params).await?;
 
         let messages = result.get("messages").and_then(|v| v.as_array()).cloned().unwrap_or_default();
-        Ok(messages.into_iter().enumerate().map(|(idx, m)| peeked_message_from_agent_json(idx, &m)).collect())
+        Ok(PeekMessagesResult::complete(
+            messages.into_iter().enumerate().map(|(idx, m)| peeked_message_from_agent_json(idx, &m)).collect(),
+        ))
     }
 
     async fn expire_messages(&self, _topic: &TopicRef, _sub: &str, _expire_seconds: i64) -> Result<(), String> {

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

@@ -160,7 +160,7 @@ pub trait MessageQueueAdmin: Send + Sync {
         sub: &str,
         count: u32,
         options: PeekMessagesOptions,
-    ) -> Result<Vec<PeekedMessage>, String>;
+    ) -> Result<PeekMessagesResult, String>;
     async fn expire_messages(&self, topic: &TopicRef, sub: &str, expire_seconds: i64) -> Result<(), String>;
 
     /// RocketMQ: read subscription group config from broker metadata.

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

@@ -476,9 +476,9 @@ pub async fn mq_peek_messages_core(
     sub: String,
     count: u32,
     options: Option<PeekMessagesOptions>,
-) -> Result<Vec<PeekedMessage>, String> {
+) -> Result<PeekMessagesResult, String> {
     if count == 0 {
-        return Ok(Vec::new());
+        return Ok(PeekMessagesResult::default());
     }
     if count > MAX_PEEK_MESSAGES {
         return Err(format!("Peek message count must be between 1 and {MAX_PEEK_MESSAGES}"));

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

@@ -496,13 +496,45 @@ pub struct PeekedMessage {
     pub payload_text: Option<String>,
 }
 
-/// Optional hints for reading messages. Pulsar ignores these today; Kafka uses
-/// them to optionally narrow a non-committing peek to one partition / offset.
-/// When omitted, Kafka peeks across all partitions from each partition's earliest
-/// readable offset (still capped by `count`).
+/// A message browse response. `incomplete` is set when the adapter could only
+/// return a partial snapshot before its configured deadline or scan limit.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PeekMessagesResult {
+    #[serde(default)]
+    pub messages: Vec<PeekedMessage>,
+    #[serde(default)]
+    pub incomplete: bool,
+}
+
+impl PeekMessagesResult {
+    pub fn complete(messages: Vec<PeekedMessage>) -> Self {
+        Self { messages, incomplete: false }
+    }
+}
+
+/// Kafka's explicit starting position for a non-committing message peek.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum PeekStartPosition {
+    Latest,
+    Earliest,
+    Offset,
+}
+
+/// Optional hints for reading messages. Non-Kafka adapters ignore
+/// `start_position` and retain their existing partition/offset handling.
+/// For Kafka, an omitted position keeps legacy behavior: it starts at the
+/// earliest available message unless an older caller supplies `offset`.
+/// `start_position: Offset` requires a non-negative offset. When no partition
+/// is supplied, Kafka reads forward from that offset in every topic partition.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PeekMessagesOptions {
+    /// Kafka's explicit read starting point. This remains optional so callers
+    /// that predate it retain their original earliest/offset behavior.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub start_position: Option<PeekStartPosition>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub partition: Option<i32>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1191,4 +1223,31 @@ mod tests {
         assert!(!caps.supports_cluster_monitoring);
         assert!(caps.supports_tenants);
     }
+
+    #[test]
+    fn peek_options_omit_optional_fields_and_serialize_start_position() {
+        let legacy: super::PeekMessagesOptions =
+            serde_json::from_str(r#"{"partition":2}"#).expect("legacy peek options");
+        assert_eq!(legacy.start_position, None);
+        assert_eq!(legacy.partition, Some(2));
+        assert_eq!(legacy.offset, None);
+
+        let latest: super::PeekMessagesOptions =
+            serde_json::from_str(r#"{"startPosition":"latest"}"#).expect("latest peek options");
+        assert_eq!(latest.start_position, Some(super::PeekStartPosition::Latest));
+        let json = serde_json::to_value(latest).expect("serialize latest peek options");
+        assert_eq!(json.get("startPosition").and_then(|value| value.as_str()), Some("latest"));
+    }
+
+    #[test]
+    fn peek_result_serializes_completion_status_and_accepts_legacy_responses() {
+        let complete = super::PeekMessagesResult::complete(Vec::new());
+        let json = serde_json::to_value(&complete).expect("serialize complete peek result");
+        assert_eq!(json, serde_json::json!({ "messages": [], "incomplete": false }));
+
+        let legacy: super::PeekMessagesResult =
+            serde_json::from_value(serde_json::json!({ "messages": [] })).expect("deserialize legacy peek result");
+        assert!(!legacy.incomplete);
+        assert!(legacy.messages.is_empty());
+    }
 }