[ISSUE #2390]🤡Implement PopMessageProcessor is_pop_should_stop method🚀

mxsm · Jan 24, 2025 · cebb22f · cebb22f
1 parent 144edb2
commit cebb22f
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/rocketmq-broker/src/processor/pop_message_processor.rs b/rocketmq-broker/src/processor/pop_message_processor.rs
@@ -1164,7 +1164,13 @@ where
         group: &CheetahString,
         queue_id: i32,
     ) -> bool {
-        unimplemented!("PopMessageProcessor is_pop_should_stop")
+        let broker_config = self.broker_runtime_inner.broker_config();
+        broker_config.enable_pop_message_threshold
+            && self
+                .broker_runtime_inner
+                .pop_inflight_message_counter()
+                .get_group_pop_in_flight_message_num(topic, group, queue_id)
+                > broker_config.pop_inflight_message_threshold
     }
 
     fn get_pop_offset(

diff --git a/rocketmq-common/src/common/broker/broker_config.rs b/rocketmq-common/src/common/broker/broker_config.rs
@@ -203,6 +203,8 @@ pub struct BrokerConfig {
     pub pop_polling_map_size: usize,
     pub max_pop_polling_size: u64,
     pub pop_polling_size: usize,
+    pub enable_pop_message_threshold: bool,
+    pub pop_inflight_message_threshold: i64,
 }
 
 impl Default for BrokerConfig {
@@ -311,6 +313,8 @@ impl Default for BrokerConfig {
             pop_polling_map_size: 100000,
             max_pop_polling_size: 100000,
             pop_polling_size: 1024,
+            enable_pop_message_threshold: false,
+            pop_inflight_message_threshold: 10000,
         }
     }
 }