Repository: mxsm/rocketmq-rust (language: rust)

Issue:
[Feature🚀] Implement PopBufferMergeService#add_ck_just_offset method
### Is Your Feature Request Related to a Problem?

_No response_

### Describe the Solution You'd Like

_No response_

### Describe Alternatives You've Considered

_No response_

### Additional Context

_No response_


Discussion / hints:

Fix patch:
diff --git a/rocketmq-broker/src/processor/pop_message_processor.rs b/rocketmq-broker/src/processor/pop_message_processor.rs
index da38dbbf..2fc76aa6 100644
--- a/rocketmq-broker/src/processor/pop_message_processor.rs
+++ b/rocketmq-broker/src/processor/pop_message_processor.rs
@@ -1047,19 +1047,22 @@ where
                                 queue_id,
                                 final_offset,
                             );
-                    } else if !self.append_check_point(
-                        request_header,
-                        topic,
-                        revive_qid,
-                        queue_id,
-                        final_offset,
-                        &result_inner,
-                        pop_time as i64,
-                        self.broker_runtime_inner
-                            .broker_config()
-                            .broker_name
-                            .as_str(),
-                    ) {
+                    } else if !self
+                        .append_check_point(
+                            request_header,
+                            topic,
+                            revive_qid,
+                            queue_id,
+                            final_offset,
+                            &result_inner,
+                            pop_time as i64,
+                            self.broker_runtime_inner
+                                .broker_config()
+                                .broker_name
+                                .as_str(),
+                        )
+                        .await
+                    {
                         self.queue_lock_manager().unlock_with_key(lock_key).await;
                         return atomic_rest_num.load(Ordering::Acquire)
                             + result_inner.message_count() as i64;
@@ -1181,7 +1184,7 @@ where
         self.queue_lock_manager().unlock_with_key(lock_key).await;
         atomic_rest_num.load(Ordering::Acquire)
     }
-    fn append_check_point(
+    async fn append_check_point(
         &self,
         request_header: &PopMessageRequestHeader,
         topic: &str,
@@ -1215,12 +1218,16 @@ where
             get_message_tmp_result.next_begin_offset(),
         ) {
             true => true,
-            false => pop_buffer_merge_service_ref_mut.add_ck_just_offset(
-                ck,
-                revive_qid,
-                -1,
-                get_message_tmp_result.next_begin_offset(),
-            ),
+            false => {
+                pop_buffer_merge_service_ref_mut
+                    .add_ck_just_offset(
+                        ck,
+                        revive_qid,
+                        -1,
+                        get_message_tmp_result.next_begin_offset(),
+                    )
+                    .await
+            }
         }
     }
 
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
index 50f1aa3f..3c1bd9b0 100644
--- a/rocketmq-broker/src/processor/processor_service/pop_buffer_merge_service.rs
+++ b/rocketmq-broker/src/processor/processor_service/pop_buffer_merge_service.rs
@@ -194,14 +194,48 @@ impl<MS: MessageStore> PopBufferMergeService<MS> {
         }
     }
 
-    pub fn add_ck_just_offset(
-        &mut self,
+    pub async fn add_ck_just_offset(
+        &self,
         point: PopCheckPoint,
         revive_queue_id: i32,
         revive_queue_offset: i64,
         next_begin_offset: i64,
     ) -> bool {
-        unimplemented!("add_ck  not implemented")
+        // Create a new wrapper with justOffset flag set to true
+        let point_wrapper = ArcMut::new(PopCheckPointWrapper::new_with_offset(
+            revive_queue_id,
+            revive_queue_offset,
+            Arc::new(point),
+            next_begin_offset,
+            true,
+        ));
+
+        // Check for merge key conflict in the buffer
+        let merge_key = point_wrapper.get_merge_key();
+        if self.buffer.contains_key(merge_key) {
+            // Log warning about merge key conflict
+            warn!(
+                "[PopBuffer]mergeKey conflict when add ckJustOffset. ck:{:?}, mergeKey:{}",
+                point_wrapper, merge_key
+            );
+            return false;
+        }
+
+        // Put checkpoint to store with condition: !checkQueueOk
+        let should_run_in_current = !self.check_queue_ok(&point_wrapper);
+        self.put_ck_to_store(point_wrapper.as_ref(), should_run_in_current)
+            .await;
+
+        // Add to offset queue
+        self.put_offset_queue(point_wrapper.clone());
+        // Log if enabled
+        if self.broker_runtime_inner.broker_config().enable_pop_log {
+            info!("[PopBuffer]add ck just offset, {:?}", point_wrapper);
+        }
+        // Add to buffer and increment counter
+        self.buffer.insert(merge_key.clone(), point_wrapper);
+        self.counter.fetch_add(1, Ordering::AcqRel);
+        true
     }
 
     pub fn add_ack(&mut self, revive_qid: i32, ack_msg: &dyn AckMessage) -> bool {


Test patch: