#810 Optimize TopicConfigAndMappingSerializeWrapper

rocketmq-broker/src/broker_runtime.rs

@@ -31,6 +31,7 @@ use rocketmq_common::ArcCellWrapper;
 use rocketmq_common::TimeUtils::get_current_millis;
 use rocketmq_common::UtilAll::compute_next_morning_time_millis;
 use rocketmq_remoting::protocol::body::topic_info_wrapper::topic_config_wrapper::TopicConfigAndMappingSerializeWrapper;
+use rocketmq_remoting::protocol::body::topic_info_wrapper::topic_config_wrapper::TopicConfigSerializeWrapper;
 use rocketmq_remoting::protocol::namespace_util::NamespaceUtil;
 use rocketmq_remoting::protocol::static_topic::topic_queue_mapping_detail::TopicQueueMappingDetail;
 use rocketmq_remoting::protocol::DataVersion;
@@ -925,7 +926,10 @@ impl BrokerRuntimeInner {
         data_version: DataVersion,
     ) {
         let mut serialize_wrapper = TopicConfigAndMappingSerializeWrapper {
-            data_version: Some(data_version),
+            topic_config_serialize_wrapper: TopicConfigSerializeWrapper {
+                data_version: data_version.clone(),
+                topic_config_table: Default::default(),
+            },
             ..Default::default()
         };
 
@@ -947,7 +951,9 @@ impl BrokerRuntimeInner {
                 register_topic_config,
             );
         }
-        serialize_wrapper.topic_config_table = Some(topic_config_table);
+        serialize_wrapper
+            .topic_config_serialize_wrapper
+            .topic_config_table = topic_config_table;
         let mut topic_queue_mapping_info_map = HashMap::new();
         for topic_config in topic_config_list {
             if let Some(ref value) = self

rocketmq-broker/src/processor/admin_broker_processor/topic_request_handler.rs

@@ -27,6 +27,7 @@ use rocketmq_remoting::code::response_code::ResponseCode;
 use rocketmq_remoting::net::channel::Channel;
 use rocketmq_remoting::protocol::body::create_topic_list_request_body::CreateTopicListRequestBody;
 use rocketmq_remoting::protocol::body::topic_info_wrapper::topic_config_wrapper::TopicConfigAndMappingSerializeWrapper;
+use rocketmq_remoting::protocol::body::topic_info_wrapper::topic_config_wrapper::TopicConfigSerializeWrapper;
 use rocketmq_remoting::protocol::header::create_topic_request_header::CreateTopicRequestHeader;
 use rocketmq_remoting::protocol::header::delete_topic_request_header::DeleteTopicRequestHeader;
 use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
@@ -365,20 +366,20 @@ impl TopicRequestHandler {
                 .data_version
                 .lock()
                 .clone(),
-            topic_config_table: Some(
-                self.inner
-                    .topic_config_manager
-                    .topic_config_table()
-                    .lock()
-                    .clone(),
-            ),
-            data_version: Some(
-                self.inner
+            topic_config_serialize_wrapper: TopicConfigSerializeWrapper {
+                data_version: self
+                    .inner
                     .topic_config_manager
                     .data_version()
                     .as_ref()
                     .clone(),
-            ),
+                topic_config_table: self
+                    .inner
+                    .topic_config_manager
+                    .topic_config_table()
+                    .lock()
+                    .clone(),
+            },
             ..Default::default()
         };
         let content = topic_config_and_mapping_serialize_wrapper.to_json();

rocketmq-broker/src/topic/manager/topic_config_manager.rs

@@ -234,7 +234,10 @@ impl TopicConfigManager {
             self.data_version.mut_from_ref().next_version();
         }
         TopicConfigAndMappingSerializeWrapper {
-            topic_config_table: Some(topic_config_table),
+            topic_config_serialize_wrapper: rocketmq_remoting::protocol::body::topic_info_wrapper::topic_config_wrapper::TopicConfigSerializeWrapper {
+                topic_config_table: topic_config_table,
+                data_version: self.data_version.as_ref().clone(),
+            },
             topic_queue_mapping_info_map,
             ..TopicConfigAndMappingSerializeWrapper::default()
         }

rocketmq-namesrv/src/route/route_info_manager.rs

@@ -94,6 +94,7 @@ impl RouteInfoManager {
         }
     }
 }
+
 //impl register broker
 impl RouteInfoManager {
     pub fn register_broker(
@@ -161,9 +162,8 @@ impl RouteInfoManager {
                 if let Some(val) = self.broker_live_table.get(&addr_info_old) {
                     let old_state_version = val.data_version().state_version();
                     let new_state_version = topic_config_serialize_wrapper
+                        .topic_config_serialize_wrapper
                         .data_version()
-                        .as_ref()
-                        .unwrap()
                         .state_version();
                     if old_state_version > new_state_version {
                         warn!(
@@ -186,15 +186,16 @@ impl RouteInfoManager {
                 }
             }
         }
-        let size = if let Some(val) = topic_config_serialize_wrapper.topic_config_table() {
-            val.len()
-        } else {
-            0
-        };
+        let size = topic_config_serialize_wrapper
+            .topic_config_serialize_wrapper
+            .topic_config_table()
+            .len();
         if !broker_data.broker_addrs().contains_key(&broker_id) && size == 1 {
             warn!(
                 "Can't register topicConfigWrapper={:?} because broker[{}]={} has not registered.",
-                topic_config_serialize_wrapper.topic_config_table(),
+                topic_config_serialize_wrapper
+                    .topic_config_serialize_wrapper
+                    .topic_config_table(),
                 broker_id,
                 broker_addr
             );
@@ -214,79 +215,77 @@ impl RouteInfoManager {
         let broker_data = broker_data.clone();
         //handle master or prime slave topic config update
         if is_master || is_prime_slave {
-            if let Some(tc_table) = topic_config_serialize_wrapper.topic_config_table() {
-                let topic_queue_mapping_info_map =
-                    topic_config_serialize_wrapper.topic_queue_mapping_info_map();
-
-                // Delete the topics that don't exist in tcTable from the current broker
-                // Static topic is not supported currently
-                if self.namesrv_config.delete_topic_with_broker_registration
-                    && topic_queue_mapping_info_map.is_empty()
-                {
-                    let old_topic_set = self.topic_set_of_broker_name(&broker_name);
-                    let new_topic_set = tc_table
-                        .keys()
-                        .map(|item| item.to_string())
-                        .collect::<HashSet<String>>();
-                    let to_delete_topics = new_topic_set
-                        .difference(&old_topic_set)
-                        .map(|item| item.to_string())
-                        .collect::<HashSet<String>>();
-                    for to_delete_topic in to_delete_topics {
-                        let queue_data_map = self.topic_queue_table.get_mut(&to_delete_topic);
-                        if let Some(queue_data) = queue_data_map {
-                            let removed_qd = queue_data.remove(&broker_name);
-                            if let Some(ref removed_qd_inner) = removed_qd {
-                                info!(
-                                    "broker[{}] delete topic[{}] queue[{:?}] because of master \
-                                     change",
-                                    broker_name, to_delete_topic, removed_qd_inner
-                                );
-                            }
-                            if queue_data.is_empty() {
-                                self.topic_queue_table.remove(&to_delete_topic);
-                            }
+            let tc_table = topic_config_serialize_wrapper
+                .topic_config_serialize_wrapper
+                .topic_config_table();
+            let topic_queue_mapping_info_map =
+                topic_config_serialize_wrapper.topic_queue_mapping_info_map();
+
+            // Delete the topics that don't exist in tcTable from the current broker
+            // Static topic is not supported currently
+            if self.namesrv_config.delete_topic_with_broker_registration
+                && topic_queue_mapping_info_map.is_empty()
+            {
+                let old_topic_set = self.topic_set_of_broker_name(&broker_name);
+                let new_topic_set = tc_table
+                    .keys()
+                    .map(|item| item.to_string())
+                    .collect::<HashSet<String>>();
+                let to_delete_topics = new_topic_set
+                    .difference(&old_topic_set)
+                    .map(|item| item.to_string())
+                    .collect::<HashSet<String>>();
+                for to_delete_topic in to_delete_topics {
+                    let queue_data_map = self.topic_queue_table.get_mut(&to_delete_topic);
+                    if let Some(queue_data) = queue_data_map {
+                        let removed_qd = queue_data.remove(&broker_name);
+                        if let Some(ref removed_qd_inner) = removed_qd {
+                            info!(
+                                "broker[{}] delete topic[{}] queue[{:?}] because of master change",
+                                broker_name, to_delete_topic, removed_qd_inner
+                            );
+                        }
+                        if queue_data.is_empty() {
+                            self.topic_queue_table.remove(&to_delete_topic);
                         }
                     }
                 }
-                let default_data_version = DataVersion::default();
-                let data_version = topic_config_serialize_wrapper
-                    .data_version()
-                    .as_ref()
-                    .unwrap_or(&default_data_version);
-                for topic_config in tc_table.values() {
-                    let mut config = topic_config.clone();
-                    if (register_first
-                        || self.is_topic_config_changed(
-                            &cluster_name,
-                            &broker_addr,
-                            data_version,
-                            &broker_name,
-                            topic_config.topic_name.as_ref().unwrap(),
-                        ))
-                        && is_prime_slave
-                        && broker_data.enable_acting_master()
-                    {
-                        config.perm &= !PermName::PERM_WRITE;
-                    }
-                    self.create_and_update_queue_data(&broker_name, config);
-                }
-                if self.is_broker_topic_config_changed(&cluster_name, &broker_addr, data_version)
-                    || register_first
+            }
+            let data_version = topic_config_serialize_wrapper
+                .topic_config_serialize_wrapper
+                .data_version();
+            for topic_config in tc_table.values() {
+                let mut config = topic_config.clone();
+                if (register_first
+                    || self.is_topic_config_changed(
+                        &cluster_name,
+                        &broker_addr,
+                        data_version,
+                        &broker_name,
+                        topic_config.topic_name.as_ref().unwrap(),
+                    ))
+                    && is_prime_slave
+                    && broker_data.enable_acting_master()
                 {
-                    for (topic, vtq_info) in topic_queue_mapping_info_map {
-                        if !self.topic_queue_mapping_info_table.contains_key(topic) {
-                            self.topic_queue_mapping_info_table
-                                .insert(topic.to_string(), HashMap::new());
-                        }
+                    config.perm &= !PermName::PERM_WRITE;
+                }
+                self.create_and_update_queue_data(&broker_name, config);
+            }
+            if self.is_broker_topic_config_changed(&cluster_name, &broker_addr, data_version)
+                || register_first
+            {
+                for (topic, vtq_info) in topic_queue_mapping_info_map {
+                    if !self.topic_queue_mapping_info_table.contains_key(topic) {
                         self.topic_queue_mapping_info_table
-                            .get_mut(topic)
-                            .unwrap()
-                            .insert(
-                                vtq_info.bname.as_ref().unwrap().to_string(),
-                                vtq_info.clone(),
-                            );
+                            .insert(topic.to_string(), HashMap::new());
                     }
+                    self.topic_queue_mapping_info_table
+                        .get_mut(topic)
+                        .unwrap()
+                        .insert(
+                            vtq_info.bname.as_ref().unwrap().to_string(),
+                            vtq_info.clone(),
+                        );
                 }
             }
         }
@@ -301,11 +300,10 @@ impl RouteInfoManager {
                     .expect("Time went backwards")
                     .as_millis() as i64,
                 DEFAULT_BROKER_CHANNEL_EXPIRED_TIME,
-                if let Some(data_version) = topic_config_serialize_wrapper.data_version() {
-                    data_version.clone()
-                } else {
-                    DataVersion::default()
-                },
+                topic_config_serialize_wrapper
+                    .topic_config_serialize_wrapper
+                    .data_version()
+                    .clone(),
                 ha_server_addr.clone(),
                 remote_addr,
             ),

rocketmq-remoting/src/protocol/body/topic_info_wrapper/topic_config_wrapper.rs

@@ -35,11 +35,16 @@ pub struct TopicConfigAndMappingSerializeWrapper {
     #[serde(rename = "mappingDataVersion")]
     pub mapping_data_version: DataVersion,
 
-    #[serde(rename = "topicConfigTable")]
-    pub topic_config_table: Option<HashMap<String, TopicConfig>>,
+    #[serde(flatten)]
+    pub topic_config_serialize_wrapper: TopicConfigSerializeWrapper,
+}
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub struct TopicConfigSerializeWrapper {
+    #[serde(rename = "topicConfigTable")]
+    pub topic_config_table: HashMap<String, TopicConfig>,
     #[serde(rename = "dataVersion")]
-    pub data_version: Option<DataVersion>,
+    pub data_version: DataVersion,
 }
 
 impl TopicConfigAndMappingSerializeWrapper {
@@ -54,12 +59,17 @@ impl TopicConfigAndMappingSerializeWrapper {
     pub fn mapping_data_version(&self) -> &DataVersion {
         &self.mapping_data_version
     }
+    pub fn topic_config_serialize_wrapper(&self) -> &TopicConfigSerializeWrapper {
+        &self.topic_config_serialize_wrapper
+    }
+}
 
-    pub fn topic_config_table(&self) -> &Option<HashMap<String, TopicConfig>> {
+impl TopicConfigSerializeWrapper {
+    pub fn topic_config_table(&self) -> &HashMap<String, TopicConfig> {
         &self.topic_config_table
     }
 
-    pub fn data_version(&self) -> &Option<DataVersion> {
+    pub fn data_version(&self) -> &DataVersion {
         &self.data_version
     }
 }
@@ -70,8 +80,16 @@ impl Default for TopicConfigAndMappingSerializeWrapper {
             topic_queue_mapping_info_map: HashMap::new(),
             topic_queue_mapping_detail_map: HashMap::new(),
             mapping_data_version: DataVersion::new(),
-            topic_config_table: None,
-            data_version: None,
+            topic_config_serialize_wrapper: TopicConfigSerializeWrapper::default(),
+        }
+    }
+}
+
+impl Default for TopicConfigSerializeWrapper {
+    fn default() -> Self {
+        Self {
+            topic_config_table: HashMap::new(),
+            data_version: DataVersion::new(),
         }
     }
 }
@@ -87,9 +105,18 @@ mod tests {
         let wrapper = TopicConfigAndMappingSerializeWrapper::default();
         assert!(wrapper.topic_queue_mapping_info_map.is_empty());
         assert!(wrapper.topic_queue_mapping_detail_map.is_empty());
-        //assert_eq!(wrapper.mapping_data_version, DataVersion::new());
-        assert!(wrapper.topic_config_table.is_none());
-        assert!(wrapper.data_version.is_none());
+        assert_eq!(wrapper.mapping_data_version, DataVersion::new());
+        assert_eq!(
+            wrapper
+                .topic_config_serialize_wrapper()
+                .topic_config_table()
+                .is_empty(),
+            true
+        );
+        assert_eq!(
+            wrapper.topic_config_serialize_wrapper().data_version(),
+            &DataVersion::new()
+        );
     }
 
     #[test]
@@ -100,8 +127,8 @@ mod tests {
         let topic_queue_mapping_detail = TopicQueueMappingDetail::default();
         let data_version = DataVersion::default();
 
-        wrapper.topic_config_table = Some(HashMap::new());
-        wrapper.data_version = Some(data_version.clone());
+        let topic_config_serialize_wrapper = TopicConfigSerializeWrapper::default();
+        wrapper.topic_config_serialize_wrapper = topic_config_serialize_wrapper.clone();
         wrapper
             .topic_queue_mapping_info_map
             .insert("test".to_string(), topic_queue_mapping_info.clone());
@@ -117,8 +144,10 @@ mod tests {
             wrapper.topic_queue_mapping_detail_map(),
             &HashMap::from([("test".to_string(), topic_queue_mapping_detail)])
         );
-        //assert_eq!(wrapper.mapping_data_version(), &DataVersion::new());
-        assert_eq!(wrapper.topic_config_table(), &Some(HashMap::new()));
-        assert_eq!(wrapper.data_version(), &Some(data_version));
+        assert_eq!(wrapper.mapping_data_version(), &data_version);
+        assert_eq!(
+            wrapper.topic_config_serialize_wrapper(),
+            &topic_config_serialize_wrapper
+        );
     }
 }