diff --git a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java b/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java
index ff29fbe8e3..f9a97f4089 100644
--- a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java
+++ b/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java
@@ -973,16 +973,7 @@ public final class KafkaAgent {
         String groupId = stringOrEmpty(params, "groupId");
         String topic = stringOrEmpty(params, "topic");
 
-        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
-        JsonArray offsetArray = params.has("offsets") && params.get("offsets").isJsonArray()
-            ? params.getAsJsonArray("offsets") : new JsonArray();
-
-        for (JsonElement el : offsetArray) {
-            JsonObject offsetObj = el.getAsJsonObject();
-            int partition = offsetObj.get("partition").getAsInt();
-            long offset = offsetObj.get("offset").getAsLong();
-            offsets.put(new TopicPartition(topic, partition), new OffsetAndMetadata(offset));
-        }
+        Map<TopicPartition, OffsetAndMetadata> offsets = explicitConsumerGroupOffsets(params, topic);
 
         // If no explicit offsets, check for a "position" parameter.
         if (offsets.isEmpty()) {
@@ -1026,6 +1017,58 @@ public final class KafkaAgent {
         return Collections.singletonMap("ok", true);
     }
 
+    static Map<TopicPartition, OffsetAndMetadata> explicitConsumerGroupOffsets(JsonObject params, String topic) {
+        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
+        if (!params.has("offsets")) {
+            return offsets;
+        }
+        JsonElement offsetsElement = params.get("offsets");
+        if (!offsetsElement.isJsonArray()) {
+            throw new IllegalArgumentException("offsets must be an array");
+        }
+        JsonArray offsetArray = offsetsElement.getAsJsonArray();
+        if (offsetArray.isEmpty()) {
+            throw new IllegalArgumentException("offsets must contain at least one partition offset");
+        }
+        for (JsonElement element : offsetArray) {
+            if (!element.isJsonObject()) {
+                throw new IllegalArgumentException("each offset must be an object");
+            }
+            JsonObject value = element.getAsJsonObject();
+            int partition = nonNegativeExactInt(value, "partition");
+            long offset = nonNegativeExactLong(value, "offset");
+            TopicPartition topicPartition = new TopicPartition(topic, partition);
+            if (offsets.put(topicPartition, new OffsetAndMetadata(offset)) != null) {
+                throw new IllegalArgumentException("duplicate partition in offsets: " + partition);
+            }
+        }
+        return offsets;
+    }
+
+    private static int nonNegativeExactInt(JsonObject object, String name) {
+        long value = nonNegativeExactLong(object, name);
+        if (value > Integer.MAX_VALUE) {
+            throw new IllegalArgumentException(name + " is outside the supported integer range");
+        }
+        return (int) value;
+    }
+
+    private static long nonNegativeExactLong(JsonObject object, String name) {
+        JsonElement element = object.get(name);
+        if (element == null || !element.isJsonPrimitive() || !element.getAsJsonPrimitive().isNumber()) {
+            throw new IllegalArgumentException(name + " must be a non-negative integer");
+        }
+        try {
+            java.math.BigDecimal decimal = element.getAsBigDecimal();
+            if (decimal.signum() < 0 || decimal.stripTrailingZeros().scale() > 0) {
+                throw new IllegalArgumentException(name + " must be a non-negative integer");
+            }
+            return decimal.longValueExact();
+        } catch (ArithmeticException error) {
+            throw new IllegalArgumentException(name + " is outside the supported integer range", error);
+        }
+    }
+
     static OffsetSpec offsetSpecForPosition(String position, Long timestampMs) {
         String normalized = position == null ? "latest" : position.trim().toLowerCase(Locale.ROOT);
         return switch (normalized) {
diff --git a/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java b/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java
index 360d079a32..8f0ec8176a 100644
--- a/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java
+++ b/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java
@@ -54,6 +54,40 @@ class KafkaAgentTest {
     @TempDir
     Path tempDir;
 
+    @Test
+    void explicitConsumerGroupOffsetsAcceptOneBoundedPartitionOffset() {
+        JsonObject params = JsonParser.parseString("""
+            {"offsets":[{"partition":1,"offset":42}]}
+            """).getAsJsonObject();
+
+        assertEquals(
+            Map.of(new TopicPartition("events", 1), new org.apache.kafka.clients.consumer.OffsetAndMetadata(42L)),
+            KafkaAgent.explicitConsumerGroupOffsets(params, "events")
+        );
+    }
+
+    @Test
+    void explicitConsumerGroupOffsetsRejectMalformedOrNegativeValues() {
+        for (String json : List.of(
+            "{\"offsets\":[{\"partition\":-1,\"offset\":42}]}",
+            "{\"offsets\":[{\"partition\":1,\"offset\":-1}]}",
+            "{\"offsets\":[{\"partition\":1.5,\"offset\":42}]}",
+            "{\"offsets\":[{\"partition\":1,\"offset\":42.5}]}",
+            "{\"offsets\":[{\"partition\":\"1\",\"offset\":42}]}",
+            "{\"offsets\":[{\"partition\":1,\"offset\":\"42\"}]}",
+            "{\"offsets\":[{\"partition\":1}]}",
+            "{\"offsets\":[42]}",
+            "{\"offsets\":[{\"partition\":2147483648,\"offset\":42}]}",
+            "{\"offsets\":[{\"partition\":1,\"offset\":9223372036854775808}]}",
+            "{\"offsets\":[{\"partition\":1,\"offset\":42},{\"partition\":1,\"offset\":43}]}",
+            "{\"offsets\":[]}",
+            "{\"offsets\":{}}"
+        )) {
+            JsonObject params = JsonParser.parseString(json).getAsJsonObject();
+            assertThrows(IllegalArgumentException.class, () -> KafkaAgent.explicitConsumerGroupOffsets(params, "events"), json);
+        }
+    }
+
     @Test
     void topicListingFallsBackWhenDescriptionsAreUnsupported() throws Exception {
         Object result = KafkaAgent.topicListResult(
diff --git a/apps/desktop/src/components/mq/SubscriptionsPanel.vue b/apps/desktop/src/components/mq/SubscriptionsPanel.vue
index 4eb12238de..b8db362771 100644
--- a/apps/desktop/src/components/mq/SubscriptionsPanel.vue
+++ b/apps/desktop/src/components/mq/SubscriptionsPanel.vue
@@ -65,8 +65,10 @@ const formData = ref({
 });
 
 const resetFormData = ref({
-  position: "latest" as "earliest" | "latest" | "timestamp",
+  position: "latest" as "earliest" | "latest" | "timestamp" | "partitionOffset",
   timestampMs: Date.now(),
+  partition: 0,
+  offset: 0,
 });
 
 const skipFormData = ref({
@@ -261,6 +263,8 @@ function openResetDialog(sub: SubscriptionInfo) {
   resetFormData.value = {
     position: "latest",
     timestampMs: Date.now(),
+    partition: 0,
+    offset: 0,
   };
   showResetDialog.value = true;
 }
@@ -365,6 +369,13 @@ async function handleResetCursor() {
     let pos: ResetPosition;
     if (resetFormData.value.position === "timestamp") {
       pos = { kind: "timestamp", timestampMs: resetFormData.value.timestampMs };
+    } else if (resetFormData.value.position === "partitionOffset") {
+      const { partition, offset } = resetFormData.value;
+      if (!Number.isSafeInteger(partition) || partition < 0 || !Number.isSafeInteger(offset) || offset < 0) {
+        error.value = t("mqSubscriptions.nonNegativeIntegerRequired");
+        return;
+      }
+      pos = { kind: "partitionOffset", partition, offset };
     } else {
       pos = { kind: resetFormData.value.position };
     }
@@ -650,6 +661,10 @@ watch(
                 <input type="radio" v-model="resetFormData.position" value="timestamp" :disabled="readOnly" />
                 {{ t("mqSubscriptions.timestamp") }}
               </label>
+              <label v-if="mqSystemKind === 'kafka'" class="radio-label">
+                <input type="radio" v-model="resetFormData.position" value="partitionOffset" :disabled="readOnly" />
+                {{ t("mqSubscriptions.partitionOffset") }}
+              </label>
             </div>
           </div>
           <div v-if="resetFormData.position === 'timestamp'" class="form-group">
@@ -657,6 +672,12 @@ watch(
             <input v-model.number="resetFormData.timestampMs" type="number" :disabled="readOnly" />
             <div class="form-hint">{{ t("mqSubscriptions.currentTime", { time: new Date(resetFormData.timestampMs).toLocaleString() }) }}</div>
           </div>
+          <div v-if="mqSystemKind === 'kafka' && resetFormData.position === 'partitionOffset'" class="form-group">
+            <label>{{ t("mqSubscriptions.partition") }}</label>
+            <input v-model.number="resetFormData.partition" data-testid="reset-partition" type="number" min="0" step="1" :disabled="readOnly" />
+            <label>{{ t("mqSubscriptions.offset") }}</label>
+            <input v-model.number="resetFormData.offset" data-testid="reset-offset" type="number" min="0" step="1" :disabled="readOnly" />
+          </div>
           <div v-if="error" class="form-error">{{ error }}</div>
         </div>
         <div class="dialog-footer">
diff --git a/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts b/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts
index 21bd1b4b68..36c18f58ce 100644
--- a/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts
+++ b/apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts
@@ -58,6 +58,12 @@ function buttonByText(container: ParentNode, text: string): HTMLButtonElement {
   return button;
 }
 
+function buttonWithExactText(container: ParentNode, text: string): HTMLButtonElement {
+  const button = [...container.querySelectorAll<HTMLButtonElement>("button")].find((item) => item.textContent?.trim() === text);
+  if (!button) throw new Error(`Button not found: ${text}`);
+  return button;
+}
+
 function deferred<T>() {
   let resolve!: (value: T) => void;
   const promise = new Promise<T>((resolvePromise) => {
@@ -75,6 +81,7 @@ async function mountPanel(overrides: Record<string, unknown> = {}) {
     tenant: "_kafka",
     namespace: "default",
     mqSystemKind: "kafka",
+    supportsResetCursor: true,
     supportsPeekMessages: true,
     ...overrides,
   });
@@ -198,6 +205,77 @@ describe("SubscriptionsPanel message peek", () => {
   });
 });
 
+describe("SubscriptionsPanel Kafka absolute offset reset", () => {
+  async function openAbsoluteReset(panel: HTMLDivElement) {
+    await vi.waitFor(() => {
+      expect([...panel.querySelectorAll("button")].some((button) => button.textContent?.includes("mqSubscriptions.resetCursor"))).toBe(true);
+    });
+    buttonByText(panel, "mqSubscriptions.resetCursor").click();
+    await flushUi();
+    const absolute = panel.querySelector<HTMLInputElement>('input[value="partitionOffset"]');
+    expect(absolute).toBeTruthy();
+    absolute!.checked = true;
+    absolute!.dispatchEvent(new Event("change", { bubbles: true }));
+    await vi.waitFor(() => {
+      expect(panel.querySelector('[data-testid="reset-partition"]')).toBeTruthy();
+    });
+  }
+
+  it("shows Kafka-only partition fields and dispatches one exact absolute offset", async () => {
+    backend.mqResetCursor.mockResolvedValue(undefined);
+    const panel = await mountPanel();
+    await openAbsoluteReset(panel);
+
+    const partition = panel.querySelector<HTMLInputElement>('[data-testid="reset-partition"]');
+    const offset = panel.querySelector<HTMLInputElement>('[data-testid="reset-offset"]');
+    expect(partition).toBeTruthy();
+    expect(offset).toBeTruthy();
+    partition!.value = "1";
+    partition!.dispatchEvent(new Event("input", { bubbles: true }));
+    offset!.value = "42";
+    offset!.dispatchEvent(new Event("input", { bubbles: true }));
+
+    buttonWithExactText(panel, "mqSubscriptions.reset").click();
+    await flushUi();
+
+    expect(backend.mqResetCursor).toHaveBeenCalledWith(
+      "mq-1",
+      expect.objectContaining({ topic: "events" }),
+      "orders-consumer",
+      { kind: "partitionOffset", partition: 1, offset: 42 },
+    );
+  });
+
+  it.each([
+    ["partition", "-1"],
+    ["partition", "1.5"],
+    ["offset", "-1"],
+    ["offset", "42.5"],
+    ["offset", "9007199254740992"],
+  ])("rejects invalid %s value %s", async (field, value) => {
+    const panel = await mountPanel();
+    await openAbsoluteReset(panel);
+    const input = panel.querySelector<HTMLInputElement>(`[data-testid="reset-${field}"]`)!;
+    input.value = value;
+    input.dispatchEvent(new Event("input", { bubbles: true }));
+    buttonWithExactText(panel, "mqSubscriptions.reset").click();
+    await vi.waitFor(() => {
+      expect(panel.textContent).toContain("mqSubscriptions.nonNegativeIntegerRequired");
+    });
+    expect(backend.mqResetCursor).not.toHaveBeenCalled();
+  });
+
+  it("hides the absolute offset option outside Kafka", async () => {
+    const pulsarPanel = await mountPanel({ mqSystemKind: "pulsar", tenant: "public" });
+    await vi.waitFor(() => {
+      expect([...pulsarPanel.querySelectorAll("button")].some((button) => button.textContent?.includes("mqSubscriptions.resetCursor"))).toBe(true);
+    });
+    buttonByText(pulsarPanel, "mqSubscriptions.resetCursor").click();
+    await flushUi();
+    expect(pulsarPanel.querySelector('input[value="partitionOffset"]')).toBeNull();
+  });
+});
+
 describe("SubscriptionsPanel RocketMQ online members", () => {
   it("shows '-' when onlineMembers is unknown before enrich completes", async () => {
     const enrich = deferred<ReturnType<typeof subscription>[]>();
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts
index d30b88caa9..8fda0485df 100644
--- a/apps/desktop/src/i18n/locales/en.ts
+++ b/apps/desktop/src/i18n/locales/en.ts
@@ -7040,6 +7040,10 @@ export default {
     latest: "Latest message (Latest)",
     timestamp: "Specific timestamp",
     timestampMs: "Timestamp (ms)",
+    partitionOffset: "Partition and absolute offset",
+    partition: "Partition",
+    offset: "Offset",
+    nonNegativeIntegerRequired: "Partition and offset must be non-negative integers",
     currentTime: "Current time: {time}",
     reset: "Reset",
     skipDialogTitle: "Skip messages: {name}",
diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts
index f709350003..1bb50bd9f4 100644
--- a/apps/desktop/src/i18n/locales/zh-CN.ts
+++ b/apps/desktop/src/i18n/locales/zh-CN.ts
@@ -6961,6 +6961,10 @@ export default withEnglishFallback({
     latest: "最新消息（Latest）",
     timestamp: "指定时间戳",
     timestampMs: "时间戳（毫秒）",
+    partitionOffset: "指定分区和绝对 Offset",
+    partition: "分区",
+    offset: "Offset",
+    nonNegativeIntegerRequired: "分区和 Offset 必须是非负整数",
     currentTime: "当前时间: {time}",
     reset: "重置",
     skipDialogTitle: "跳过消息: {name}",
diff --git a/apps/desktop/src/types/mq.ts b/apps/desktop/src/types/mq.ts
index 4412656d8f..619243c88e 100644
--- a/apps/desktop/src/types/mq.ts
+++ b/apps/desktop/src/types/mq.ts
@@ -232,7 +232,12 @@ export interface ProducerInfo {
   clientVersion: string;
 }
 
-export type ResetPosition = { kind: "earliest" } | { kind: "latest" } | { kind: "timestamp"; timestampMs: number } | { kind: "messageId"; ledgerId: number; entryId: number };
+export type ResetPosition =
+  | { kind: "earliest" }
+  | { kind: "latest" }
+  | { kind: "timestamp"; timestampMs: number }
+  | { kind: "partitionOffset"; partition: number; offset: number }
+  | { kind: "messageId"; ledgerId: number; entryId: number };
 
 export type SkipCount = { kind: "all" } | { kind: "count"; count: number };
 
diff --git a/crates/dbx-core/src/mq/adapters/kafka.rs b/crates/dbx-core/src/mq/adapters/kafka.rs
index 2bef343f7c..5ccac3b907 100644
--- a/crates/dbx-core/src/mq/adapters/kafka.rs
+++ b/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!({
diff --git a/crates/dbx-core/src/mq/adapters/pulsar.rs b/crates/dbx-core/src/mq/adapters/pulsar.rs
index c311f5b011..cac27425a4 100644
--- a/crates/dbx-core/src/mq/adapters/pulsar.rs
+++ b/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]
diff --git a/crates/dbx-core/src/mq/adapters/rocketmq.rs b/crates/dbx-core/src/mq/adapters/rocketmq.rs
index ae32e5d08c..5895afe0fc 100644
--- a/crates/dbx-core/src/mq/adapters/rocketmq.rs
+++ b/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": [] });
diff --git a/crates/dbx-core/src/mq/types.rs b/crates/dbx-core/src/mq/types.rs
index 500f5aa538..e8cdaa32f7 100644
--- a/crates/dbx-core/src/mq/types.rs
+++ b/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")]