[ISSUE #1763]🚀Implement ConsumerOrderInfoManager#auto_clean method function🔥

mxsm · mxsm · commit 13eae183f9d3 · 2024-12-14T13:29:03.000+08:00
diff --git a/rocketmq-broker/src/broker_runtime.rs b/rocketmq-broker/src/broker_runtime.rs
@@ -103,7 +103,7 @@ pub(crate) struct BrokerRuntime {
     #[cfg(feature = "local_file_store")]
     subscription_group_manager: Arc<SubscriptionGroupManager<DefaultMessageStore>>,
     consumer_filter_manager: Arc<ConsumerFilterManager>,
-    consumer_order_info_manager: Arc<ConsumerOrderInfoManager>,
+    consumer_order_info_manager: Arc<ConsumerOrderInfoManager<DefaultMessageStore>>,
     #[cfg(feature = "local_file_store")]
     message_store: Option<ArcMut<DefaultMessageStore>>,
     #[cfg(feature = "local_file_store")]
@@ -154,7 +154,7 @@ impl Clone for BrokerRuntime {
             consumer_offset_manager: self.consumer_offset_manager.clone(),
             subscription_group_manager: self.subscription_group_manager.clone(),
             consumer_filter_manager: Arc::new(Default::default()),
-            consumer_order_info_manager: Arc::new(Default::default()),
+            consumer_order_info_manager: self.consumer_order_info_manager.clone(),
             message_store: self.message_store.clone(),
             broker_stats: self.broker_stats.clone(),
             schedule_message_service: self.schedule_message_service.clone(),
@@ -241,19 +241,23 @@ impl BrokerRuntime {
             topic_route_info_manager.clone(),
             broker_outer_api.clone(),
         ));
+        let subscription_group_manager =
+            Arc::new(SubscriptionGroupManager::new(broker_config.clone(), None));
+        let consumer_order_info_manager = Arc::new(ConsumerOrderInfoManager::new(
+            broker_config.clone(),
+            Arc::new(topic_config_manager.clone()),
+            subscription_group_manager.clone(),
+        ));
         Self {
             broker_config: broker_config.clone(),
             message_store_config,
             server_config,
             topic_config_manager,
             topic_queue_mapping_manager,
             consumer_offset_manager: ConsumerOffsetManager::new(broker_config.clone(), None),
-            subscription_group_manager: Arc::new(SubscriptionGroupManager::new(
-                broker_config.clone(),
-                None,
-            )),
+            subscription_group_manager,
             consumer_filter_manager: Arc::new(Default::default()),
-            consumer_order_info_manager: Arc::new(Default::default()),
+            consumer_order_info_manager,
             message_store: None,
             broker_stats: None,
             schedule_message_service: Default::default(),
diff --git a/rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs b/rocketmq-broker/src/offset/manager/consumer_order_info_manager.rs
@@ -17,6 +17,7 @@
 
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::fmt::Display;
 use std::ops::Deref;
 use std::sync::Arc;
 
@@ -27,24 +28,46 @@ use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use rocketmq_common::TimeUtils::get_current_millis;
 use serde::Deserialize;
 use serde::Serialize;
+use tracing::info;
 use tracing::warn;
 
 use crate::broker_path_config_helper::get_consumer_order_info_path;
 use crate::offset::manager::consumer_order_info_lock_manager::ConsumerOrderInfoLockManager;
+use crate::subscription::manager::subscription_group_manager::SubscriptionGroupManager;
+use crate::topic::manager::topic_config_manager::TopicConfigManager;
 
 const TOPIC_GROUP_SEPARATOR: &str = "@";
 const CLEAN_SPAN_FROM_LAST: u64 = 24 * 3600 * 1000;
 
-#[derive(Default)]
-pub(crate) struct ConsumerOrderInfoManager {
+pub(crate) struct ConsumerOrderInfoManager<MS> {
     pub(crate) broker_config: Arc<BrokerConfig>,
     pub(crate) consumer_order_info_wrapper: parking_lot::Mutex<ConsumerOrderInfoWrapper>,
     pub(crate) consumer_order_info_lock_manager: Option<ConsumerOrderInfoLockManager>,
+    pub(crate) topic_config_manager: Arc<TopicConfigManager>,
+    pub(crate) subscription_group_manager: Arc<SubscriptionGroupManager<MS>>,
+}
+
+impl<MS> ConsumerOrderInfoManager<MS> {
+    pub fn new(
+        broker_config: Arc<BrokerConfig>,
+        topic_config_manager: Arc<TopicConfigManager>,
+        subscription_group_manager: Arc<SubscriptionGroupManager<MS>>,
+    ) -> ConsumerOrderInfoManager<MS> {
+        Self {
+            broker_config,
+            consumer_order_info_wrapper: parking_lot::Mutex::new(
+                ConsumerOrderInfoWrapper::default(),
+            ),
+            consumer_order_info_lock_manager: None,
+            topic_config_manager,
+            subscription_group_manager,
+        }
+    }
 }
 
 //Fully implemented will be removed
 #[allow(unused_variables)]
-impl ConfigManager for ConsumerOrderInfoManager {
+impl<MS> ConfigManager for ConsumerOrderInfoManager<MS> {
     fn config_file_path(&self) -> String {
         get_consumer_order_info_path(self.broker_config.store_path_root_dir.as_str())
     }
@@ -81,8 +104,92 @@ impl ConfigManager for ConsumerOrderInfoManager {
     }
 }
 
-impl ConsumerOrderInfoManager {
-    fn auto_clean(&self) {}
+impl<MS> ConsumerOrderInfoManager<MS> {
+    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;
+
+        // Iterate through the `table` entries (topic@group -> queueId -> OrderInfo)
+        let mut keys_to_remove = Vec::new(); // Temporary storage for keys to remove
+
+        for (topic_at_group, qs) in table.iter_mut() {
+            let arrays: Vec<&str> = topic_at_group.split('@').collect();
+            if arrays.len() != 2 {
+                continue;
+            }
+            let topic = arrays[0];
+            let group = arrays[1];
+
+            let topic_config = self
+                .topic_config_manager
+                .select_topic_config(&CheetahString::from(topic));
+            if topic_config.is_none() {
+                info!(
+                    "Topic not exist, Clean order info, {}:{:?}",
+                    topic_at_group, qs
+                );
+                keys_to_remove.push(topic_at_group.clone());
+                continue;
+            }
+            let subscription_group_wrapper = self
+                .subscription_group_manager
+                .subscription_group_wrapper()
+                .lock();
+            let subscription_group_config = subscription_group_wrapper
+                .subscription_group_table()
+                .get(&CheetahString::from(group));
+            if subscription_group_config.is_none() {
+                info!(
+                    "Group not exist, Clean order info, {}:{:?}",
+                    topic_at_group, qs
+                );
+                keys_to_remove.push(topic_at_group.clone());
+                continue;
+            }
+
+            if qs.is_empty() {
+                info!(
+                    "Order table is empty, Clean order info, {}:{:?}",
+                    topic_at_group, qs
+                );
+                keys_to_remove.push(topic_at_group.clone());
+                continue;
+            }
+            let topic_config = topic_config.unwrap();
+            // Clean individual queues in the current topic@group
+            let mut queues_to_remove = Vec::new();
+            for (queue_id, order_info) in qs.iter_mut() {
+                if *queue_id == topic_config.read_queue_nums as i32 {
+                    queues_to_remove.push(*queue_id);
+                    info!(
+                        "Queue not exist, Clean order info, {}:{}, {}",
+                        topic_at_group, order_info, topic_config
+                    );
+                    continue;
+                }
+            }
+
+            // Remove stale or invalid queues
+            for queue_id in queues_to_remove {
+                qs.remove(&queue_id);
+                info!(
+                    "Removed queue {} for topic@group {}",
+                    queue_id, topic_at_group
+                );
+            }
+
+            // If all queues are removed, mark topic@group for removal
+            if qs.is_empty() {
+                keys_to_remove.push(topic_at_group.clone());
+            }
+        }
+
+        // Now, remove all topics/groups from the table that need to be cleaned
+        for key in keys_to_remove {
+            table.remove(&key);
+            info!("Removed topic@group {}", key);
+        }
+    }
 
     pub fn update_next_visible_time(
         &self,
@@ -167,6 +274,25 @@ pub(crate) struct OrderInfo {
     attempt_id: String,
 }
 
+impl Display for OrderInfo {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "OrderInfo {{ popTime: {}, invisibleTime: {:?}, offsetList: {:?}, \
+             offsetNextVisibleTime: {:?}, offsetConsumedCount: {:?}, lastConsumeTimestamp: {}, \
+             commitOffsetBit: {}, attemptId: {} }}",
+            self.pop_time,
+            self.invisible_time,
+            self.offset_list,
+            self.offset_next_visible_time,
+            self.offset_consumed_count,
+            self.last_consume_timestamp,
+            self.commit_offset_bit,
+            self.attempt_id
+        )
+    }
+}
+
 impl OrderInfo {
     /// Builds a list of offsets from a given list of queue offsets.
     /// If the list contains only one element, it returns the same list.
diff --git a/rocketmq-broker/src/processor/change_invisible_time_processor.rs b/rocketmq-broker/src/processor/change_invisible_time_processor.rs
@@ -59,7 +59,7 @@ pub struct ChangeInvisibleTimeProcessor<MS> {
     topic_config_manager: TopicConfigManager,
     message_store: ArcMut<MS>,
     consumer_offset_manager: Arc<ConsumerOffsetManager>,
-    consumer_order_info_manager: Arc<ConsumerOrderInfoManager>,
+    consumer_order_info_manager: Arc<ConsumerOrderInfoManager<MS>>,
     broker_stats_manager: Arc<BrokerStatsManager>,
     pop_buffer_merge_service: ArcMut<PopBufferMergeService>,
     escape_bridge: ArcMut<EscapeBridge<MS>>,
@@ -74,7 +74,7 @@ impl<MS> ChangeInvisibleTimeProcessor<MS> {
         topic_config_manager: TopicConfigManager,
         message_store: ArcMut<MS>,
         consumer_offset_manager: Arc<ConsumerOffsetManager>,
-        consumer_order_info_manager: Arc<ConsumerOrderInfoManager>,
+        consumer_order_info_manager: Arc<ConsumerOrderInfoManager<MS>>,
         broker_stats_manager: Arc<BrokerStatsManager>,
         pop_buffer_merge_service: ArcMut<PopBufferMergeService>,
         escape_bridge: ArcMut<EscapeBridge<MS>>,
diff --git a/rocketmq-broker/src/subscription/manager/subscription_group_manager.rs b/rocketmq-broker/src/subscription/manager/subscription_group_manager.rs
@@ -38,7 +38,7 @@ pub const TOPIC_MAX_LENGTH: usize = 127;
 
 pub(crate) struct SubscriptionGroupManager<MS> {
     pub(crate) broker_config: Arc<BrokerConfig>,
-    subscription_group_wrapper: Arc<parking_lot::Mutex<SubscriptionGroupWrapper>>,
+    pub(crate) subscription_group_wrapper: Arc<parking_lot::Mutex<SubscriptionGroupWrapper>>,
     pub(crate) message_store: Option<MS>,
 }
 
@@ -55,6 +55,10 @@ impl<MS> SubscriptionGroupManager<MS> {
             message_store,
         }
     }
+
+    pub fn subscription_group_wrapper(&self) -> &Arc<parking_lot::Mutex<SubscriptionGroupWrapper>> {
+        &self.subscription_group_wrapper
+    }
 }
 
 impl<MS> ConfigManager for SubscriptionGroupManager<MS> {
@@ -199,7 +203,7 @@ where
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-struct SubscriptionGroupWrapper {
+pub(crate) struct SubscriptionGroupWrapper {
     subscription_group_table: HashMap<CheetahString, SubscriptionGroupConfig>,
     forbidden_table: HashMap<CheetahString, HashMap<CheetahString, i32>>,
     data_version: DataVersion,
diff --git a/rocketmq-common/src/common/config.rs b/rocketmq-common/src/common/config.rs
@@ -16,6 +16,7 @@
  */
 
 use std::collections::HashMap;
+use std::fmt::Display;
 
 use cheetah_string::CheetahString;
 use serde::Deserialize;
@@ -39,6 +40,24 @@ pub struct TopicConfig {
     pub attributes: HashMap<CheetahString, CheetahString>,
 }
 
+impl Display for TopicConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "TopicConfig {{ topic_name: {:?}, read_queue_nums: {}, write_queue_nums: {}, perm: \
+             {}, topic_filter_type: {}, topic_sys_flag: {}, order: {}, attributes: {:?} }}",
+            self.topic_name,
+            self.read_queue_nums,
+            self.write_queue_nums,
+            self.perm,
+            self.topic_filter_type,
+            self.topic_sys_flag,
+            self.order,
+            self.attributes
+        )
+    }
+}
+
 impl Default for TopicConfig {
     fn default() -> Self {
         Self {

Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ pub const TOPIC_MAX_LENGTH: usize = 127;`
`38`	`38`
`39`	`39`	`pub(crate) struct SubscriptionGroupManager<MS> {`
`40`	`40`	`pub(crate) broker_config: Arc<BrokerConfig>,`
`41`		`- subscription_group_wrapper: Arc<parking_lot::Mutex<SubscriptionGroupWrapper>>,`
	`41`	`+ pub(crate) subscription_group_wrapper: Arc<parking_lot::Mutex<SubscriptionGroupWrapper>>,`
`42`	`42`	`pub(crate) message_store: Option<MS>,`
`43`	`43`	`}`
`44`	`44`
`@@ -55,6 +55,10 @@ impl<MS> SubscriptionGroupManager<MS> {`
`55`	`55`	`message_store,`
`56`	`56`	`}`
`57`	`57`	`}`
	`58`	`+`
	`59`	`+ pub fn subscription_group_wrapper(&self) -> &Arc<parking_lot::Mutex<SubscriptionGroupWrapper>> {`
	`60`	`+ &self.subscription_group_wrapper`
	`61`	`+ }`
`58`	`62`	`}`
`59`	`63`
`60`	`64`	`impl<MS> ConfigManager for SubscriptionGroupManager<MS> {`
`@@ -199,7 +203,7 @@ where`
`199`	`203`
`200`	`204`	`#[derive(Debug, Clone, Default, Serialize, Deserialize)]`
`201`	`205`	`#[serde(rename_all = "camelCase")]`
`202`		`-struct SubscriptionGroupWrapper {`
	`206`	`+pub(crate) struct SubscriptionGroupWrapper {`
`203`	`207`	`subscription_group_table: HashMap<CheetahString, SubscriptionGroupConfig>,`
`204`	`208`	`forbidden_table: HashMap<CheetahString, HashMap<CheetahString, i32>>,`
`205`	`209`	`data_version: DataVersion,`