kafrust_protocol/api/
describe_log_dirs.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 35;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct DescribeLogDirsRequest {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub topics: Option<Vec<DescribeLogDirsTopic>>,
12}
13
14impl DescribeLogDirsRequest {
15    pub fn encode_v1(&self) -> Result<Vec<u8>> {
16        let mut encoder = Encoder::new();
17        RequestHeader {
18            api_key: API_KEY,
19            api_version: 1,
20            correlation_id: self.correlation_id,
21            client_id: self.client_id.clone(),
22        }
23        .encode_v1(&mut encoder)?;
24        encode_legacy_topics(&mut encoder, self.topics.as_deref())?;
25        Ok(encoder.into_bytes())
26    }
27
28    pub fn encode_v2(&self, api_version: i16) -> Result<Vec<u8>> {
29        let mut encoder = Encoder::new();
30        RequestHeader {
31            api_key: API_KEY,
32            api_version,
33            correlation_id: self.correlation_id,
34            client_id: self.client_id.clone(),
35        }
36        .encode_v2(&mut encoder)?;
37        encoder.write_compact_array(self.topics.as_deref(), |encoder, topic| {
38            encoder.write_compact_string(&topic.name)?;
39            encoder.write_compact_array(Some(&topic.partition_indexes), |encoder, partition| {
40                encoder.write_i32(*partition);
41                Ok(())
42            })?;
43            encoder.write_empty_tagged_fields();
44            Ok(())
45        })?;
46        encoder.write_empty_tagged_fields();
47        Ok(encoder.into_bytes())
48    }
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct DescribeLogDirsTopic {
53    pub name: String,
54    pub partition_indexes: Vec<i32>,
55}
56
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct DescribeLogDirsResponse {
59    pub throttle_time_ms: i32,
60    pub error_code: i16,
61    pub results: Vec<DescribeLogDirsResult>,
62}
63
64impl DescribeLogDirsResponse {
65    pub fn decode_body_v1(decoder: &mut Decoder<'_>) -> Result<Self> {
66        decode_body(decoder, false, false, false, false)
67    }
68
69    pub fn decode_body_v2(decoder: &mut Decoder<'_>) -> Result<Self> {
70        decode_body(decoder, true, false, false, false)
71    }
72
73    pub fn decode_body_v3(decoder: &mut Decoder<'_>) -> Result<Self> {
74        decode_body(decoder, true, true, false, false)
75    }
76
77    pub fn decode_body_v4(decoder: &mut Decoder<'_>) -> Result<Self> {
78        decode_body(decoder, true, true, true, false)
79    }
80
81    pub fn decode_body_v5(decoder: &mut Decoder<'_>) -> Result<Self> {
82        decode_body(decoder, true, true, true, true)
83    }
84}
85
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct DescribeLogDirsResult {
88    pub error_code: i16,
89    pub log_dir: String,
90    pub topics: Vec<DescribeLogDirsTopicResult>,
91    pub total_bytes: i64,
92    pub usable_bytes: i64,
93    pub is_cordoned: bool,
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub struct DescribeLogDirsTopicResult {
98    pub name: String,
99    pub partitions: Vec<DescribeLogDirsPartitionResult>,
100}
101
102#[derive(Debug, Clone, PartialEq, Eq)]
103pub struct DescribeLogDirsPartitionResult {
104    pub partition_index: i32,
105    pub partition_size: i64,
106    pub offset_lag: i64,
107    pub is_future: bool,
108}
109
110fn encode_legacy_topics(
111    encoder: &mut Encoder,
112    topics: Option<&[DescribeLogDirsTopic]>,
113) -> Result<()> {
114    encoder.write_array(topics, |encoder, topic| {
115        encoder.write_string(&topic.name)?;
116        encoder.write_array(Some(&topic.partition_indexes), |encoder, partition| {
117            encoder.write_i32(*partition);
118            Ok(())
119        })
120    })
121}
122
123fn decode_body(
124    decoder: &mut Decoder<'_>,
125    flexible: bool,
126    has_error_code: bool,
127    has_capacity: bool,
128    has_cordoned: bool,
129) -> Result<DescribeLogDirsResponse> {
130    let throttle_time_ms = decoder.read_i32()?;
131    let error_code = if has_error_code {
132        decoder.read_i16()?
133    } else {
134        0
135    };
136    let results = if flexible {
137        decoder
138            .read_compact_array("describe log dirs results", |decoder| {
139                decode_flexible_result(decoder, has_capacity, has_cordoned)
140            })?
141            .unwrap_or_default()
142    } else {
143        decoder
144            .read_array("describe log dirs results", |decoder| {
145                decode_legacy_result(decoder, has_capacity, has_cordoned)
146            })?
147            .unwrap_or_default()
148    };
149    if flexible {
150        decoder.read_tagged_fields()?;
151    }
152    Ok(DescribeLogDirsResponse {
153        throttle_time_ms,
154        error_code,
155        results,
156    })
157}
158
159fn decode_legacy_result(
160    decoder: &mut Decoder<'_>,
161    has_capacity: bool,
162    has_cordoned: bool,
163) -> Result<DescribeLogDirsResult> {
164    let result = DescribeLogDirsResult {
165        error_code: decoder.read_i16()?,
166        log_dir: decoder.read_string()?,
167        topics: decoder
168            .read_array("describe log dirs topics", decode_legacy_topic_result)?
169            .unwrap_or_default(),
170        total_bytes: if has_capacity {
171            decoder.read_i64()?
172        } else {
173            -1
174        },
175        usable_bytes: if has_capacity {
176            decoder.read_i64()?
177        } else {
178            -1
179        },
180        is_cordoned: if has_cordoned {
181            decoder.read_bool()?
182        } else {
183            false
184        },
185    };
186    Ok(result)
187}
188
189fn decode_legacy_topic_result(decoder: &mut Decoder<'_>) -> Result<DescribeLogDirsTopicResult> {
190    Ok(DescribeLogDirsTopicResult {
191        name: decoder.read_string()?,
192        partitions: decoder
193            .read_array("describe log dirs partitions", decode_partition_result)?
194            .unwrap_or_default(),
195    })
196}
197
198fn decode_flexible_result(
199    decoder: &mut Decoder<'_>,
200    has_capacity: bool,
201    has_cordoned: bool,
202) -> Result<DescribeLogDirsResult> {
203    let result = DescribeLogDirsResult {
204        error_code: decoder.read_i16()?,
205        log_dir: decoder.read_compact_string()?,
206        topics: decoder
207            .read_compact_array("describe log dirs topics", decode_flexible_topic_result)?
208            .unwrap_or_default(),
209        total_bytes: if has_capacity {
210            decoder.read_i64()?
211        } else {
212            -1
213        },
214        usable_bytes: if has_capacity {
215            decoder.read_i64()?
216        } else {
217            -1
218        },
219        is_cordoned: if has_cordoned {
220            decoder.read_bool()?
221        } else {
222            false
223        },
224    };
225    decoder.read_tagged_fields()?;
226    Ok(result)
227}
228
229fn decode_flexible_topic_result(decoder: &mut Decoder<'_>) -> Result<DescribeLogDirsTopicResult> {
230    let result = DescribeLogDirsTopicResult {
231        name: decoder.read_compact_string()?,
232        partitions: decoder
233            .read_compact_array(
234                "describe log dirs partitions",
235                decode_flexible_partition_result,
236            )?
237            .unwrap_or_default(),
238    };
239    decoder.read_tagged_fields()?;
240    Ok(result)
241}
242
243fn decode_flexible_partition_result(
244    decoder: &mut Decoder<'_>,
245) -> Result<DescribeLogDirsPartitionResult> {
246    let result = decode_partition_result(decoder)?;
247    decoder.read_tagged_fields()?;
248    Ok(result)
249}
250
251fn decode_partition_result(decoder: &mut Decoder<'_>) -> Result<DescribeLogDirsPartitionResult> {
252    Ok(DescribeLogDirsPartitionResult {
253        partition_index: decoder.read_i32()?,
254        partition_size: decoder.read_i64()?,
255        offset_lag: decoder.read_i64()?,
256        is_future: decoder.read_bool()?,
257    })
258}
259
260#[cfg(test)]
261#[allow(clippy::unwrap_used)]
262mod tests {
263    use super::{DescribeLogDirsRequest, DescribeLogDirsResponse, DescribeLogDirsTopic, API_KEY};
264    use crate::codec::{Decoder, Encoder};
265
266    #[test]
267    fn encodes_describe_log_dirs_v1_request_with_nullable_topics() {
268        let request = DescribeLogDirsRequest {
269            correlation_id: 35,
270            client_id: Some("kafrust".to_owned()),
271            topics: None,
272        };
273
274        let bytes = request.encode_v1().unwrap();
275        assert_eq!(&bytes[0..4], &[0, API_KEY as u8, 0, 1]);
276        assert_eq!(&bytes[4..8], &[0, 0, 0, 35]);
277        assert_eq!(&bytes[17..21], &[255, 255, 255, 255]);
278    }
279
280    #[test]
281    fn encodes_describe_log_dirs_v2_request_with_partition_filter() {
282        let request = DescribeLogDirsRequest {
283            correlation_id: 36,
284            client_id: None,
285            topics: Some(vec![DescribeLogDirsTopic {
286                name: "orders".to_owned(),
287                partition_indexes: vec![0, 2],
288            }]),
289        };
290
291        let bytes = request.encode_v2(2).unwrap();
292        assert_eq!(&bytes[0..4], &[0, API_KEY as u8, 0, 2]);
293        assert_eq!(&bytes[4..8], &[0, 0, 0, 36]);
294        assert!(bytes
295            .windows(7)
296            .any(|window| { window == [7, b'o', b'r', b'd', b'e', b'r', b's'] }));
297        assert!(bytes.ends_with(&[0]));
298    }
299
300    #[test]
301    fn decodes_describe_log_dirs_v5_response_with_capacity_and_tags() {
302        let mut bytes = Encoder::new();
303        bytes.write_i32(11);
304        bytes.write_i16(0);
305        bytes.write_unsigned_varint(2); // one result
306        bytes.write_i16(0);
307        bytes.write_compact_string("/var/lib/kafka").unwrap();
308        bytes.write_unsigned_varint(2); // one topic
309        bytes.write_compact_string("orders").unwrap();
310        bytes.write_unsigned_varint(2); // one partition
311        bytes.write_i32(0);
312        bytes.write_i64(4096);
313        bytes.write_i64(3);
314        bytes.write_bool(false);
315        bytes.write_empty_tagged_fields();
316        bytes.write_empty_tagged_fields();
317        bytes.write_i64(1_000_000);
318        bytes.write_i64(900_000);
319        bytes.write_bool(false);
320        bytes.write_empty_tagged_fields();
321        bytes.write_empty_tagged_fields();
322        let encoded = bytes.into_bytes();
323        let mut decoder = Decoder::new(&encoded);
324
325        let response = DescribeLogDirsResponse::decode_body_v5(&mut decoder).unwrap();
326
327        assert_eq!(response.throttle_time_ms, 11);
328        assert_eq!(response.results[0].log_dir, "/var/lib/kafka");
329        assert_eq!(response.results[0].topics[0].name, "orders");
330        assert_eq!(
331            response.results[0].topics[0].partitions[0].partition_size,
332            4096
333        );
334        assert_eq!(response.results[0].topics[0].partitions[0].offset_lag, 3);
335        assert_eq!(response.results[0].total_bytes, 1_000_000);
336        assert_eq!(response.results[0].usable_bytes, 900_000);
337        assert!(!response.results[0].is_cordoned);
338        assert!(decoder.is_empty());
339    }
340
341    #[test]
342    fn decodes_describe_log_dirs_v5_response_with_multiple_directories() {
343        let mut bytes = Encoder::new();
344        bytes.write_i32(0);
345        bytes.write_i16(0);
346        bytes.write_unsigned_varint(3); // two log directories
347        for (path, total, usable, cordoned) in [
348            ("/var/lib/kafka", 100_i64, 90_i64, false),
349            ("/var/lib/kafka-2", 200_i64, 180_i64, true),
350        ] {
351            bytes.write_i16(0);
352            bytes.write_compact_string(path).unwrap();
353            bytes.write_unsigned_varint(1); // no topics
354            bytes.write_i64(total);
355            bytes.write_i64(usable);
356            bytes.write_bool(cordoned);
357            bytes.write_empty_tagged_fields();
358        }
359        bytes.write_empty_tagged_fields();
360        let encoded = bytes.into_bytes();
361        let mut decoder = Decoder::new(&encoded);
362        let response = DescribeLogDirsResponse::decode_body_v5(&mut decoder).unwrap();
363
364        assert_eq!(response.results.len(), 2);
365        assert_eq!(response.results[0].log_dir, "/var/lib/kafka");
366        assert_eq!(response.results[0].total_bytes, 100);
367        assert_eq!(response.results[1].log_dir, "/var/lib/kafka-2");
368        assert_eq!(response.results[1].usable_bytes, 180);
369        assert!(response.results[1].is_cordoned);
370        assert!(decoder.is_empty());
371    }
372}
kafrust_protocol/api/describe_log_dirs.rs

kafrust_protocol/api/
describe_log_dirs.rs
