[ISSUE #1996]🍻Implement PopMessageProcessor#reset_pop_offset method l…

…ogic🚀
mxsm · Jan 1, 2025 · c618f82 · c618f82
1 parent 59fd6b7
commit c618f82
Show file tree

Hide file tree

Showing 3 changed files with 40 additions and 3 deletions.
diff --git a/rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs b/rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs
@@ -106,6 +106,10 @@ impl<MS> ConfigManager for ConsumerOrderInfoManager<MS> {
 }
 
 impl<MS> ConsumerOrderInfoManager<MS> {
+    pub fn clear_block(&self, topic: &CheetahString, group: &CheetahString, queue_id: i32) {
+        unimplemented!()
+    }
+
     pub fn auto_clean(&self) {
         let mut consumer_order_info_wrapper = self.consumer_order_info_wrapper.lock();
         let table = &mut consumer_order_info_wrapper.table;

diff --git a/rocketmq-broker/src/processor/pop_message_processor.rs b/rocketmq-broker/src/processor/pop_message_processor.rs
@@ -1028,13 +1028,42 @@ where
                 topic,
                 queue_id,
                 offset,
-            )
+            );
         }
         offset
     }
 
-    fn reset_pop_offset(&self, topic: &str, group: &str, queue_id: i32) -> Option<i64> {
-        unimplemented!("PopMessageProcessor reset_pop_offset")
+    fn reset_pop_offset(
+        &self,
+        topic: &CheetahString,
+        group: &CheetahString,
+        queue_id: i32,
+    ) -> Option<i64> {
+        let lock_key = format!(
+            "{}{}{}{}{}",
+            topic,
+            PopAckConstants::SPLIT,
+            group,
+            PopAckConstants::SPLIT,
+            queue_id
+        );
+        let reset_offset = self
+            .consumer_offset_manager
+            .query_then_erase_reset_offset(group, topic, queue_id);
+        if reset_offset.is_some() {
+            self.consumer_order_info_manager
+                .clear_block(topic, group, queue_id);
+            self.pop_buffer_merge_service
+                .clear_offset_queue(lock_key.as_ref());
+            self.consumer_offset_manager.commit_offset(
+                "ResetPopOffset".into(),
+                group,
+                topic,
+                queue_id,
+                *reset_offset.as_ref().unwrap(),
+            )
+        }
+        reset_offset
     }
 
     pub fn queue_lock_manager(&self) -> &QueueLockManager {

diff --git a/rocketmq-broker/src/processor/processor_service/pop_buffer_merge_service.rs b/rocketmq-broker/src/processor/processor_service/pop_buffer_merge_service.rs
@@ -27,4 +27,8 @@ impl PopBufferMergeService {
     pub fn get_latest_offset(&self, _lock_key: &str) -> i64 {
         unimplemented!("Not implemented yet");
     }
+
+    pub fn clear_offset_queue(&self, _lock_key: &str) {
+        unimplemented!("Not implemented yet");
+    }
 }