[ISSUE #2004]🍻Implement PopBufferMergeService-1 🚀 (#2022)

Author: mxsm

rocketmq-broker/src/broker_runtime.rs

@@ -76,7 +76,6 @@ use crate::processor::default_pull_message_result_handler::DefaultPullMessageRes
 use crate::processor::end_transaction_processor::EndTransactionProcessor;
 use crate::processor::pop_inflight_message_counter::PopInflightMessageCounter;
 use crate::processor::pop_message_processor::PopMessageProcessor;
-use crate::processor::processor_service::pop_buffer_merge_service::PopBufferMergeService;
 use crate::processor::pull_message_processor::PullMessageProcessor;
 use crate::processor::pull_message_result_handler::PullMessageResultHandler;
 use crate::processor::query_assignment_processor::QueryAssignmentProcessor;
@@ -597,7 +596,6 @@ impl BrokerRuntime {
                 Arc::new(self.consumer_offset_manager.clone()),
                 self.consumer_order_info_manager.clone(),
                 self.broker_stats_manager.clone(),
-                ArcMut::new(PopBufferMergeService),
                 self.escape_bridge.clone(),
                 pop_message_processor,
             )),

rocketmq-broker/src/processor/ack_message_processor.rs

@@ -57,13 +57,11 @@ use crate::offset::manager::consumer_offset_manager::ConsumerOffsetManager;
 use crate::offset::manager::consumer_order_info_manager::ConsumerOrderInfoManager;
 use crate::processor::pop_inflight_message_counter::PopInflightMessageCounter;
 use crate::processor::pop_message_processor::PopMessageProcessor;
-use crate::processor::processor_service::pop_buffer_merge_service::PopBufferMergeService;
 use crate::topic::manager::topic_config_manager::TopicConfigManager;
 
 pub struct AckMessageProcessor<MS> {
     topic_config_manager: TopicConfigManager,
     message_store: ArcMut<MS>,
-    pop_buffer_merge_service: ArcMut<PopBufferMergeService>,
     escape_bridge: ArcMut<EscapeBridge<MS>>,
     store_host: SocketAddr,
     pop_inflight_message_counter: Arc<PopInflightMessageCounter>,
@@ -90,8 +88,6 @@ where
         AckMessageProcessor {
             topic_config_manager,
             message_store,
-            /* need to implement PopBufferMergeService */
-            pop_buffer_merge_service: ArcMut::new(PopBufferMergeService),
             escape_bridge,
             store_host,
             pop_inflight_message_counter,
@@ -377,7 +373,8 @@ where
         ack_msg.set_pop_time(pop_time);
         ack_msg.set_broker_name(broker_name);
         if self
-            .pop_buffer_merge_service
+            .pop_message_processor
+            .pop_buffer_merge_service_mut()
             .add_ack(r_qid, ack_msg.as_ref())
         {
             return;

rocketmq-broker/src/processor/change_invisible_time_processor.rs

@@ -51,7 +51,6 @@ use crate::failover::escape_bridge::EscapeBridge;
 use crate::offset::manager::consumer_offset_manager::ConsumerOffsetManager;
 use crate::offset::manager::consumer_order_info_manager::ConsumerOrderInfoManager;
 use crate::processor::pop_message_processor::PopMessageProcessor;
-use crate::processor::processor_service::pop_buffer_merge_service::PopBufferMergeService;
 use crate::topic::manager::topic_config_manager::TopicConfigManager;
 
 pub struct ChangeInvisibleTimeProcessor<MS> {
@@ -61,7 +60,6 @@ pub struct ChangeInvisibleTimeProcessor<MS> {
     consumer_offset_manager: Arc<ConsumerOffsetManager>,
     consumer_order_info_manager: Arc<ConsumerOrderInfoManager<MS>>,
     broker_stats_manager: Arc<BrokerStatsManager>,
-    pop_buffer_merge_service: ArcMut<PopBufferMergeService>,
     escape_bridge: ArcMut<EscapeBridge<MS>>,
     revive_topic: CheetahString,
     store_host: SocketAddr,
@@ -76,7 +74,6 @@ impl<MS> ChangeInvisibleTimeProcessor<MS> {
         consumer_offset_manager: Arc<ConsumerOffsetManager>,
         consumer_order_info_manager: Arc<ConsumerOrderInfoManager<MS>>,
         broker_stats_manager: Arc<BrokerStatsManager>,
-        pop_buffer_merge_service: ArcMut<PopBufferMergeService>,
         escape_bridge: ArcMut<EscapeBridge<MS>>,
         pop_message_processor: ArcMut<PopMessageProcessor<MS>>,
     ) -> Self {
@@ -93,7 +90,6 @@ impl<MS> ChangeInvisibleTimeProcessor<MS> {
             consumer_offset_manager,
             consumer_order_info_manager,
             broker_stats_manager,
-            pop_buffer_merge_service,
             escape_bridge,
             revive_topic: CheetahString::from_string(revive_topic),
             store_host,
@@ -275,7 +271,11 @@ where
             request_header.topic.as_str(),
             1,
         );
-        if self.pop_buffer_merge_service.add_ack(rq_id, &ack_msg) {
+        if self
+            .pop_message_processor
+            .pop_buffer_merge_service_mut()
+            .add_ack(rq_id, &ack_msg)
+        {
             return Ok(());
         }
         let mut inner = MessageExtBrokerInner::default();

rocketmq-broker/src/processor/pop_message_processor.rs

@@ -96,6 +96,7 @@ pub struct PopMessageProcessor<MS> {
     pop_buffer_merge_service: ArcMut<PopBufferMergeService>,
     pop_inflight_message_counter: Arc<PopInflightMessageCounter>,
     queue_lock_manager: QueueLockManager,
+    revive_topic: CheetahString,
 }
 
 impl<MS> PopMessageProcessor<MS> {
@@ -111,6 +112,10 @@ impl<MS> PopMessageProcessor<MS> {
         consumer_filter_manager: Arc<ConsumerFilterManager>,
         pop_inflight_message_counter: Arc<PopInflightMessageCounter>,
     ) -> Self {
+        let revive_topic = CheetahString::from_string(PopAckConstants::build_cluster_revive_topic(
+            broker_config.broker_identity.broker_cluster_name.as_str(),
+        ));
+        let queue_lock_manager = QueueLockManager::new();
         PopMessageProcessor {
             consumer_offset_manager,
             consumer_manager,
@@ -123,9 +128,13 @@ impl<MS> PopMessageProcessor<MS> {
             consumer_filter_manager,
             ck_message_number: Default::default(),
             pop_long_polling_service: ArcMut::new(PopLongPollingService),
-            pop_buffer_merge_service: ArcMut::new(PopBufferMergeService),
+            pop_buffer_merge_service: ArcMut::new(PopBufferMergeService::new(
+                revive_topic.clone(),
+                queue_lock_manager.clone(),
+            )),
             pop_inflight_message_counter,
-            queue_lock_manager: QueueLockManager::new(),
+            queue_lock_manager,
+            revive_topic,
         }
     }
 }
@@ -1160,6 +1169,14 @@ impl<MS> PopMessageProcessor<MS> {
             PopAckConstants::CK_TAG
         )
     }
+
+    pub fn pop_buffer_merge_service(&self) -> &ArcMut<PopBufferMergeService> {
+        &self.pop_buffer_merge_service
+    }
+
+    pub fn pop_buffer_merge_service_mut(&mut self) -> &mut ArcMut<PopBufferMergeService> {
+        &mut self.pop_buffer_merge_service
+    }
 }
 
 struct TimedLock {
@@ -1202,6 +1219,7 @@ impl TimedLock {
     }
 }
 
+#[derive(Clone)]
 pub struct QueueLockManager {
     expired_local_cache: Arc<Mutex<HashMap<CheetahString, TimedLock>>>,
 }

rocketmq-broker/src/processor/processor_service/pop_buffer_merge_service.rs

@@ -21,12 +21,53 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use cheetah_string::CheetahString;
+use dashmap::DashMap;
 use rocketmq_common::common::pop_ack_constants::PopAckConstants;
 use rocketmq_common::TimeUtils::get_current_millis;
 use rocketmq_store::pop::pop_check_point::PopCheckPoint;
 use rocketmq_store::pop::AckMessage;
 
-pub(crate) struct PopBufferMergeService;
+use crate::processor::pop_message_processor::QueueLockManager;
+
+pub(crate) struct PopBufferMergeService {
+    buffer: DashMap<CheetahString /* mergeKey */, PopCheckPointWrapper>,
+    commit_offsets:
+        DashMap<CheetahString /* topic@cid@queueId */, QueueWithTime<PopCheckPointWrapper>>,
+    serving: AtomicBool,
+    counter: AtomicI32,
+    scan_times: i32,
+    revive_topic: CheetahString,
+    queue_lock_manager: QueueLockManager,
+    interval: u64,
+    minute5: u64,
+    count_of_minute1: u64,
+    count_of_second1: u64,
+    count_of_second30: u64,
+    batch_ack_index_list: Vec<u8>,
+    master: AtomicBool,
+}
+
+impl PopBufferMergeService {
+    pub fn new(revive_topic: CheetahString, queue_lock_manager: QueueLockManager) -> Self {
+        let interval = 5;
+        Self {
+            buffer: DashMap::with_capacity(1024 * 16),
+            commit_offsets: DashMap::with_capacity(1024),
+            serving: AtomicBool::new(true),
+            counter: AtomicI32::new(0),
+            scan_times: 0,
+            revive_topic,
+            queue_lock_manager,
+            interval,
+            minute5: 5 * 60 * 1000,
+            count_of_minute1: 60 * 1000 / interval,
+            count_of_second1: 1000 / interval,
+            count_of_second30: 30 * 1000 / interval,
+            batch_ack_index_list: Vec::with_capacity(32),
+            master: AtomicBool::new(false),
+        }
+    }
+}
 
 impl PopBufferMergeService {
     pub fn add_ack(&mut self, _revive_qid: i32, _ack_msg: &dyn AckMessage) -> bool {