rocketmq_remoting/protocol/static_topic/
topic_queue_mapping_detail.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashMap;
16
17use rocketmq_rust::ArcMut;
18use serde::Deserialize;
19use serde::Serialize;
20
21use crate::protocol::static_topic::i32_key_map_serde;
22use crate::protocol::static_topic::logic_queue_mapping_item::LogicQueueMappingItem;
23use crate::protocol::static_topic::topic_queue_mapping_info::TopicQueueMappingInfo;
24
25#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
26pub struct TopicQueueMappingDetail {
27    #[serde(flatten)]
28    pub topic_queue_mapping_info: TopicQueueMappingInfo,
29
30    #[serde(rename = "hostedQueues")]
31    #[serde(
32        default = "default_hosted_queues",
33        deserialize_with = "i32_key_map_serde::deserialize_optional_i32_key_map"
34    )]
35    pub hosted_queues: Option<HashMap<i32 /* global id */, Vec<LogicQueueMappingItem>>>,
36}
37
38fn default_hosted_queues() -> Option<HashMap<i32, Vec<LogicQueueMappingItem>>> {
39    Some(HashMap::new())
40}
41
42impl Default for TopicQueueMappingDetail {
43    fn default() -> Self {
44        Self {
45            topic_queue_mapping_info: TopicQueueMappingInfo::default(),
46            hosted_queues: default_hosted_queues(),
47        }
48    }
49}
50
51impl TopicQueueMappingDetail {
52    pub fn get_mapping_info(
53        mapping_detail: &TopicQueueMappingDetail,
54        global_id: i32,
55    ) -> Option<&Vec<LogicQueueMappingItem>> {
56        mapping_detail.hosted_queues.as_ref()?.get(&global_id)
57    }
58
59    pub fn compute_max_offset_from_mapping(mapping_detail: &TopicQueueMappingDetail, global_id: Option<i32>) -> i64 {
60        let Some(global_id) = global_id else {
61            return -1;
62        };
63        match Self::get_mapping_info(mapping_detail, global_id) {
64            Some(mapping_items) => {
65                if mapping_items.is_empty() {
66                    return -1;
67                }
68                mapping_items
69                    .last()
70                    .map_or(-1, LogicQueueMappingItem::compute_max_static_queue_offset)
71            }
72            None => -1,
73        }
74    }
75}
76
77//impl static methods(Like java static method)
78impl TopicQueueMappingDetail {
79    pub fn build_id_map(mapping_detail: &TopicQueueMappingDetail) -> HashMap<i32, i32> {
80        let mut curr_id_map = HashMap::new();
81        let broker_name = mapping_detail.topic_queue_mapping_info.bname.as_ref();
82        if let (Some(hosted_queues), Some(broker_name)) = (&mapping_detail.hosted_queues, broker_name) {
83            for (global_id, items) in hosted_queues {
84                if let Some(last_item) = items.last() {
85                    if last_item.bname.as_ref() == Some(broker_name) {
86                        curr_id_map.insert(*global_id, last_item.queue_id);
87                    }
88                }
89            }
90        }
91        curr_id_map
92    }
93
94    pub fn clone_as_mapping_info(mapping_detail: &TopicQueueMappingDetail) -> TopicQueueMappingInfo {
95        TopicQueueMappingInfo {
96            topic: mapping_detail.topic_queue_mapping_info.topic.clone(),
97            scope: mapping_detail.topic_queue_mapping_info.scope.clone(),
98            total_queues: mapping_detail.topic_queue_mapping_info.total_queues,
99            bname: mapping_detail.topic_queue_mapping_info.bname.clone(),
100            epoch: mapping_detail.topic_queue_mapping_info.epoch,
101            dirty: mapping_detail.topic_queue_mapping_info.dirty,
102            curr_id_map: Some(Self::build_id_map(mapping_detail)),
103        }
104    }
105    pub fn put_mapping_info(
106        mut mapping_detail: ArcMut<TopicQueueMappingDetail>,
107        global_id: i32,
108        mapping_info: Vec<LogicQueueMappingItem>,
109    ) {
110        if mapping_info.is_empty() {
111            return;
112        }
113        mapping_detail
114            .hosted_queues
115            .get_or_insert_with(HashMap::new)
116            .insert(global_id, mapping_info);
117    }
118
119    pub fn check_if_as_physical(mapping_detail: &TopicQueueMappingDetail, global_id: i32) -> bool {
120        match Self::get_mapping_info(mapping_detail, global_id) {
121            None => true,
122            Some(mapping_items) => {
123                mapping_items.len() == 1
124                    && mapping_items
125                        .first()
126                        .is_some_and(|mapping_item| mapping_item.logic_offset == 0)
127            }
128        }
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135
136    #[test]
137    fn topic_queue_mapping_detail_default() {
138        let detail = TopicQueueMappingDetail::default();
139        assert_eq!(detail.topic_queue_mapping_info, TopicQueueMappingInfo::default());
140        assert_eq!(detail.hosted_queues, Some(HashMap::new()));
141    }
142
143    #[test]
144    fn topic_queue_mapping_detail_serde() {
145        let mut detail = TopicQueueMappingDetail::default();
146        detail.topic_queue_mapping_info.topic = Some("test".into());
147        let mut hosted_queues = HashMap::new();
148        hosted_queues.insert(1, vec![LogicQueueMappingItem::default()]);
149        detail.hosted_queues = Some(hosted_queues);
150
151        let json = serde_json::to_string(&detail).unwrap();
152        let expected = r#"{"topic":"test","scope":"__global__","totalQueues":0,"bname":null,"epoch":0,"dirty":false,"currIdMap":null,"hostedQueues":{"1":[{"gen":0,"queueId":0,"bname":null,"logicOffset":0,"startOffset":0,"endOffset":-1,"timeOfStart":-1,"timeOfEnd":-1}]}}"#;
153        assert_eq!(json, expected);
154
155        let deserialized: TopicQueueMappingDetail = serde_json::from_str(&json).unwrap();
156        assert_eq!(detail, deserialized);
157    }
158
159    #[test]
160    fn topic_queue_mapping_detail_missing_hosted_queues_uses_java_empty_default() {
161        let json = r#"{"topic":"test","scope":"__global__","totalQueues":0,"bname":null,"epoch":0,"dirty":false,"currIdMap":null}"#;
162
163        let detail: TopicQueueMappingDetail = serde_json::from_str(json).unwrap();
164
165        assert_eq!(detail.hosted_queues, Some(HashMap::new()));
166    }
167
168    #[test]
169    fn get_mapping_info() {
170        let mut detail = TopicQueueMappingDetail::default();
171        let items = vec![LogicQueueMappingItem::default()];
172        let mut hosted_queues = HashMap::new();
173        hosted_queues.insert(1, items.clone());
174        detail.hosted_queues = Some(hosted_queues);
175
176        assert_eq!(TopicQueueMappingDetail::get_mapping_info(&detail, 1).unwrap(), &items);
177        assert!(TopicQueueMappingDetail::get_mapping_info(&detail, 2).is_none());
178    }
179
180    #[test]
181    fn compute_max_offset_from_mapping() {
182        let mut detail = TopicQueueMappingDetail::default();
183        let mut items = vec![LogicQueueMappingItem::default()];
184        items[0].logic_offset = 100;
185        items[0].start_offset = 50;
186        items[0].end_offset = 150;
187        // max = logic_offset + end_offset - start_offset = 100 + 150 - 50 = 200
188
189        let mut hosted_queues = HashMap::new();
190        hosted_queues.insert(1, items);
191        detail.hosted_queues = Some(hosted_queues);
192
193        assert_eq!(
194            TopicQueueMappingDetail::compute_max_offset_from_mapping(&detail, Some(1)),
195            200
196        );
197
198        // Update end_offset to be invalid (less than start_offset)
199        detail.hosted_queues.as_mut().unwrap().get_mut(&1).unwrap()[0].end_offset = 20;
200        // max should be logic_offset = 100
201        assert_eq!(
202            TopicQueueMappingDetail::compute_max_offset_from_mapping(&detail, Some(1)),
203            100
204        );
205
206        // non-existent global_id
207        assert_eq!(
208            TopicQueueMappingDetail::compute_max_offset_from_mapping(&detail, Some(2)),
209            -1
210        );
211        assert_eq!(
212            TopicQueueMappingDetail::compute_max_offset_from_mapping(&detail, None),
213            -1
214        );
215    }
216
217    #[test]
218    fn clone_as_mapping_info() {
219        let detail = TopicQueueMappingDetail {
220            topic_queue_mapping_info: TopicQueueMappingInfo {
221                topic: Some("test_topic".into()),
222                total_queues: 8,
223                bname: Some("broker_a".into()),
224                epoch: 12345,
225                ..TopicQueueMappingInfo::default()
226            },
227            ..Default::default()
228        };
229
230        let info = TopicQueueMappingDetail::clone_as_mapping_info(&detail);
231        assert_eq!(info.topic, detail.topic_queue_mapping_info.topic);
232        assert_eq!(info.total_queues, detail.topic_queue_mapping_info.total_queues);
233        assert_eq!(info.bname, detail.topic_queue_mapping_info.bname);
234        assert_eq!(info.epoch, detail.topic_queue_mapping_info.epoch);
235    }
236
237    #[test]
238    fn clone_as_mapping_info_builds_current_id_map_for_leader_items() {
239        let broker_a = cheetah_string::CheetahString::from_static_str("broker_a");
240        let broker_b = cheetah_string::CheetahString::from_static_str("broker_b");
241        let mut hosted_queues = HashMap::new();
242        hosted_queues.insert(
243            0,
244            vec![LogicQueueMappingItem {
245                queue_id: 3,
246                bname: Some(broker_a.clone()),
247                ..Default::default()
248            }],
249        );
250        hosted_queues.insert(
251            1,
252            vec![LogicQueueMappingItem {
253                queue_id: 7,
254                bname: Some(broker_b),
255                ..Default::default()
256            }],
257        );
258        let detail = TopicQueueMappingDetail {
259            topic_queue_mapping_info: TopicQueueMappingInfo {
260                topic: Some("test_topic".into()),
261                total_queues: 2,
262                bname: Some(broker_a),
263                epoch: 10,
264                ..TopicQueueMappingInfo::default()
265            },
266            hosted_queues: Some(hosted_queues),
267        };
268
269        let info = TopicQueueMappingDetail::clone_as_mapping_info(&detail);
270
271        let curr_id_map = info.curr_id_map.unwrap();
272        assert_eq!(curr_id_map.get(&0), Some(&3));
273        assert!(!curr_id_map.contains_key(&1));
274    }
275
276    #[test]
277    fn put_mapping_info() {
278        let detail = ArcMut::new(TopicQueueMappingDetail::default());
279
280        let items = vec![LogicQueueMappingItem::default()];
281        TopicQueueMappingDetail::put_mapping_info(detail.clone(), 1, items.clone());
282
283        assert_eq!(detail.hosted_queues.as_ref().unwrap().get(&1).unwrap(), &items);
284
285        TopicQueueMappingDetail::put_mapping_info(detail.clone(), 2, vec![]);
286        assert!(detail.hosted_queues.as_ref().unwrap().get(&2).is_none());
287    }
288
289    #[test]
290    fn put_mapping_info_initializes_missing_hosted_queues_like_java() {
291        let detail = ArcMut::new(TopicQueueMappingDetail {
292            hosted_queues: None,
293            ..Default::default()
294        });
295
296        let items = vec![LogicQueueMappingItem::default()];
297        TopicQueueMappingDetail::put_mapping_info(detail.clone(), 1, items.clone());
298
299        assert_eq!(detail.hosted_queues.as_ref().unwrap().get(&1), Some(&items));
300    }
301
302    #[test]
303    fn check_if_as_physical_matches_java_rules() {
304        let mut detail = TopicQueueMappingDetail::default();
305        assert!(TopicQueueMappingDetail::check_if_as_physical(&detail, 0));
306
307        detail.hosted_queues.as_mut().unwrap().insert(
308            0,
309            vec![LogicQueueMappingItem {
310                logic_offset: 0,
311                ..Default::default()
312            }],
313        );
314        assert!(TopicQueueMappingDetail::check_if_as_physical(&detail, 0));
315
316        detail.hosted_queues.as_mut().unwrap().insert(
317            1,
318            vec![LogicQueueMappingItem {
319                logic_offset: 10,
320                ..Default::default()
321            }],
322        );
323        assert!(!TopicQueueMappingDetail::check_if_as_physical(&detail, 1));
324
325        detail.hosted_queues.as_mut().unwrap().insert(
326            2,
327            vec![
328                LogicQueueMappingItem {
329                    logic_offset: 0,
330                    ..Default::default()
331                },
332                LogicQueueMappingItem {
333                    logic_offset: 0,
334                    ..Default::default()
335                },
336            ],
337        );
338        assert!(!TopicQueueMappingDetail::check_if_as_physical(&detail, 2));
339    }
340}
rocketmq_remoting/protocol/static_topic/topic_queue_mapping_detail.rs

rocketmq_remoting/protocol/static_topic/
topic_queue_mapping_detail.rs
