feat(kafka): reset one partition to an absolute offset

Closes #5960

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

@@ -734,6 +734,16 @@ fn reset_cursor_params(topic: &TopicRef, sub: &str, pos: ResetPosition) -> Resul
             "position": "timestamp",
             "timestampMs": timestamp_ms,
         })),
+        ResetPosition::PartitionOffset { partition, offset } => {
+            if partition < 0 || offset < 0 {
+                return Err("Kafka partition and offset must be non-negative integers".to_string());
+            }
+            Ok(serde_json::json!({
+                "groupId": sub,
+                "topic": topic.topic,
+                "offsets": [{ "partition": partition, "offset": offset }],
+            }))
+        }
         ResetPosition::MessageId { .. } => Err("Kafka does not support cursor reset by Pulsar message id".to_string()),
     }
 }
@@ -1071,6 +1081,50 @@ mod tests {
         assert!(err.contains("message id"));
     }
 
+    #[test]
+    fn reset_cursor_params_maps_one_absolute_partition_offset() {
+        let topic = TopicRef {
+            tenant: "_kafka".to_string(),
+            namespace: "_kafka".to_string(),
+            topic: "events".to_string(),
+            persistent: true,
+            partitioned: Some(true),
+            message_type: None,
+            ..TopicRef::default()
+        };
+
+        let params =
+            reset_cursor_params(&topic, "group-a", ResetPosition::PartitionOffset { partition: 2, offset: 41 })
+                .expect("Kafka should support an absolute offset for one partition");
+
+        assert_eq!(
+            params,
+            serde_json::json!({
+                "groupId": "group-a",
+                "topic": "events",
+                "offsets": [{ "partition": 2, "offset": 41 }]
+            })
+        );
+    }
+
+    #[test]
+    fn reset_cursor_params_rejects_negative_absolute_positions() {
+        let topic = TopicRef {
+            tenant: "_kafka".to_string(),
+            namespace: "_kafka".to_string(),
+            topic: "events".to_string(),
+            persistent: true,
+            partitioned: Some(true),
+            message_type: None,
+            ..TopicRef::default()
+        };
+
+        assert!(reset_cursor_params(&topic, "group-a", ResetPosition::PartitionOffset { partition: -1, offset: 41 },)
+            .is_err());
+        assert!(reset_cursor_params(&topic, "group-a", ResetPosition::PartitionOffset { partition: 1, offset: -1 },)
+            .is_err());
+    }
+
     #[test]
     fn kafka_subscription_for_topic_includes_offline_group_with_committed_offsets() {
         let desc = serde_json::json!({

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

@@ -406,9 +406,12 @@ impl MessageQueueAdmin for PulsarAdmin {
     }
 
     async fn create_subscription(&self, topic: &TopicRef, sub: &str, pos: ResetPosition) -> Result<(), String> {
+        if matches!(pos, ResetPosition::PartitionOffset { .. }) {
+            return Err("Pulsar does not support cursor reset by Kafka partition offset".to_string());
+        }
         // PUT .../subscription/{sub} with the desired message id in the body.
         let path = self.profile.subscription_path(topic.domain(), &topic.path(), sub);
-        let body = reset_position_message_id(&pos);
+        let body = reset_position_message_id(&pos)?;
         self.send_empty(reqwest::Method::PUT, &path, Some(&body), None).await
     }
 
@@ -442,9 +445,12 @@ impl MessageQueueAdmin for PulsarAdmin {
             ResetPosition::MessageId { .. } | ResetPosition::Earliest | ResetPosition::Latest => {
                 // POST .../resetcursor with a message id body.
                 let path = self.profile.subscription_reset_cursor_path(topic.domain(), &topic.path(), sub);
-                let body = reset_position_message_id(&pos);
+                let body = reset_position_message_id(&pos)?;
                 self.send_empty(reqwest::Method::POST, &path, Some(&body), None).await
             }
+            ResetPosition::PartitionOffset { .. } => {
+                Err("Pulsar does not support cursor reset by Kafka partition offset".to_string())
+            }
         }
     }
 
@@ -1001,15 +1007,18 @@ fn is_partition_member(full: &str, partitioned: &std::collections::HashSet<Strin
 /// Build a message-id body for reset-cursor / create-subscription based on the
 /// requested position. Earliest/latest use the sentinel ledger/entry ids Pulsar
 /// recognises.
-fn reset_position_message_id(pos: &ResetPosition) -> serde_json::Value {
+fn reset_position_message_id(pos: &ResetPosition) -> Result<serde_json::Value, String> {
     match pos {
-        ResetPosition::Earliest => serde_json::json!({ "ledgerId": -1, "entryId": -1 }),
-        ResetPosition::Latest => serde_json::json!({ "ledgerId": i64::MAX, "entryId": i64::MAX }),
+        ResetPosition::Earliest => Ok(serde_json::json!({ "ledgerId": -1, "entryId": -1 })),
+        ResetPosition::Latest => Ok(serde_json::json!({ "ledgerId": i64::MAX, "entryId": i64::MAX })),
         ResetPosition::MessageId { ledger_id, entry_id } => {
-            serde_json::json!({ "ledgerId": ledger_id, "entryId": entry_id })
+            Ok(serde_json::json!({ "ledgerId": ledger_id, "entryId": entry_id }))
+        }
+        ResetPosition::PartitionOffset { .. } => {
+            Err("Pulsar does not support cursor reset by Kafka partition offset".to_string())
         }
         // Timestamp is handled by a dedicated path; default to latest here.
-        ResetPosition::Timestamp { .. } => serde_json::json!({ "ledgerId": i64::MAX, "entryId": i64::MAX }),
+        ResetPosition::Timestamp { .. } => Ok(serde_json::json!({ "ledgerId": i64::MAX, "entryId": i64::MAX })),
     }
 }
 
@@ -1614,11 +1623,12 @@ mod tests {
     #[test]
     fn reset_position_message_ids() {
         assert_eq!(
-            reset_position_message_id(&ResetPosition::Earliest),
+            reset_position_message_id(&ResetPosition::Earliest).unwrap(),
             serde_json::json!({"ledgerId": -1, "entryId": -1})
         );
-        let mid = reset_position_message_id(&ResetPosition::MessageId { ledger_id: 5, entry_id: 9 });
+        let mid = reset_position_message_id(&ResetPosition::MessageId { ledger_id: 5, entry_id: 9 }).unwrap();
         assert_eq!(mid, serde_json::json!({"ledgerId": 5, "entryId": 9}));
+        assert!(reset_position_message_id(&ResetPosition::PartitionOffset { partition: 0, offset: 9 }).is_err());
     }
 
     #[test]

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

@@ -871,6 +871,9 @@ fn reset_cursor_params(topic: &TopicRef, sub: &str, pos: ResetPosition) -> Resul
             "position": "timestamp",
             "timestampMs": timestamp_ms,
         })),
+        ResetPosition::PartitionOffset { .. } => {
+            Err("RocketMQ does not support cursor reset by Kafka partition offset".to_string())
+        }
         ResetPosition::MessageId { .. } => {
             Err("RocketMQ does not support cursor reset by Pulsar message id".to_string())
         }
@@ -1292,6 +1295,23 @@ mod tests {
         assert_eq!(params.get("position").and_then(|v| v.as_str()), Some("timestamp"));
     }
 
+    #[test]
+    fn reset_cursor_params_rejects_kafka_partition_offsets() {
+        let topic = TopicRef {
+            tenant: "_rocketmq".to_string(),
+            namespace: "default".to_string(),
+            topic: "events".to_string(),
+            persistent: true,
+            partitioned: None,
+            message_type: None,
+            ..TopicRef::default()
+        };
+
+        let error = reset_cursor_params(&topic, "group-a", ResetPosition::PartitionOffset { partition: 0, offset: 9 })
+            .expect_err("RocketMQ must not reinterpret Kafka partition offsets");
+        assert!(error.contains("Kafka partition offset"));
+    }
+
     #[test]
     fn rocketmq_subscription_for_topic_includes_offline_group_with_committed_offsets() {
         let desc = serde_json::json!({ "groupId": "orders-service", "members": [] });

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

@@ -465,10 +465,26 @@ pub enum ResetPosition {
     Latest,
     /// A specific point in time (milliseconds since epoch).
     Timestamp { timestamp_ms: i64 },
+    /// An absolute Kafka offset for one partition.
+    PartitionOffset { partition: i32, offset: i64 },
     /// A specific message id.
     MessageId { ledger_id: i64, entry_id: i64 },
 }
 
+#[cfg(test)]
+mod reset_position_tests {
+    use super::ResetPosition;
+
+    #[test]
+    fn partition_offset_uses_the_frontend_camel_case_wire_shape() {
+        let position = ResetPosition::PartitionOffset { partition: 3, offset: 27 };
+        let value = serde_json::to_value(&position).expect("serialize reset position");
+        assert_eq!(value, serde_json::json!({ "kind": "partitionOffset", "partition": 3, "offset": 27 }));
+        let decoded: ResetPosition = serde_json::from_value(value).expect("deserialize reset position");
+        assert!(matches!(decoded, ResetPosition::PartitionOffset { partition: 3, offset: 27 }));
+    }
+}
+
 /// How many messages to skip on a subscription.
 #[derive(Debug, Clone, Copy, Serialize, Deserialize)]
 #[serde(tag = "kind", rename_all = "camelCase", rename_all_fields = "camelCase")]