kafrust_protocol/api/
offset_fetch.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 9;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct OffsetFetchRequestV2 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub group_id: String,
12    pub topics: Option<Vec<OffsetFetchTopic>>,
13}
14
15impl OffsetFetchRequestV2 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 2,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_string(&self.group_id)?;
26        encoder.write_array(self.topics.as_deref(), |encoder, topic| {
27            topic.encode(encoder)
28        })?;
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct OffsetFetchTopic {
35    pub name: String,
36    pub partition_indexes: Vec<i32>,
37}
38
39impl OffsetFetchTopic {
40    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
41        encoder.write_string(&self.name)?;
42        encoder.write_array(
43            Some(self.partition_indexes.as_slice()),
44            |encoder, partition| {
45                encoder.write_i32(*partition);
46                Ok(())
47            },
48        )
49    }
50}
51
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct OffsetFetchResponseV2 {
54    pub topics: Vec<OffsetFetchTopicResponse>,
55    pub error_code: i16,
56}
57
58impl OffsetFetchResponseV2 {
59    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
60        Ok(Self {
61            topics: decoder
62                .read_array(
63                    "offset fetch topic responses",
64                    OffsetFetchTopicResponse::decode,
65                )?
66                .unwrap_or_default(),
67            error_code: decoder.read_i16()?,
68        })
69    }
70}
71
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub struct OffsetFetchTopicResponse {
74    pub name: String,
75    pub partitions: Vec<OffsetFetchPartitionResponse>,
76}
77
78impl OffsetFetchTopicResponse {
79    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
80        Ok(Self {
81            name: decoder.read_string()?,
82            partitions: decoder
83                .read_array(
84                    "offset fetch partition responses",
85                    OffsetFetchPartitionResponse::decode,
86                )?
87                .unwrap_or_default(),
88        })
89    }
90}
91
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub struct OffsetFetchPartitionResponse {
94    pub partition_index: i32,
95    pub committed_offset: i64,
96    pub metadata: Option<String>,
97    pub error_code: i16,
98}
99
100impl OffsetFetchPartitionResponse {
101    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
102        Ok(Self {
103            partition_index: decoder.read_i32()?,
104            committed_offset: decoder.read_i64()?,
105            metadata: decoder.read_nullable_string()?,
106            error_code: decoder.read_i16()?,
107        })
108    }
109}
110
111#[cfg(test)]
112#[allow(clippy::unwrap_used)]
113mod tests {
114    use super::{
115        OffsetFetchPartitionResponse, OffsetFetchRequestV2, OffsetFetchResponseV2,
116        OffsetFetchTopic, OffsetFetchTopicResponse,
117    };
118    use crate::codec::{Decoder, Encoder};
119
120    #[test]
121    fn encodes_offset_fetch_v2_request_for_partitions() {
122        let request = OffsetFetchRequestV2 {
123            correlation_id: 29,
124            client_id: Some("kafrust".to_owned()),
125            group_id: "orders-group".to_owned(),
126            topics: Some(vec![OffsetFetchTopic {
127                name: "orders".to_owned(),
128                partition_indexes: vec![0, 1],
129            }]),
130        };
131
132        assert_eq!(
133            request.encode().unwrap(),
134            [
135                0, 9, // api key
136                0, 2, // api version
137                0, 0, 0, 29, // correlation id
138                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
139                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
140                b'p', // group id
141                0, 0, 0, 1, // topic count
142                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic
143                0, 0, 0, 2, // partition count
144                0, 0, 0, 0, // partition 0
145                0, 0, 0, 1, // partition 1
146            ]
147        );
148    }
149
150    #[test]
151    fn encodes_offset_fetch_v2_request_for_all_topics() {
152        let request = OffsetFetchRequestV2 {
153            correlation_id: 31,
154            client_id: None,
155            group_id: "orders-group".to_owned(),
156            topics: None,
157        };
158
159        assert_eq!(
160            request.encode().unwrap(),
161            [
162                0, 9, // api key
163                0, 2, // api version
164                0, 0, 0, 31, // correlation id
165                0xff, 0xff, // null client id
166                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
167                b'p', // group id
168                0xff, 0xff, 0xff, 0xff, // null topics
169            ]
170        );
171    }
172
173    #[test]
174    fn decodes_offset_fetch_v2_response() {
175        let mut bytes = Encoder::new();
176        bytes.write_i32(1);
177        bytes.write_string("orders").unwrap();
178        bytes.write_i32(1);
179        bytes.write_i32(0);
180        bytes.write_i64(42);
181        bytes.write_nullable_string(Some("processed")).unwrap();
182        bytes.write_i16(0);
183        bytes.write_i16(0);
184        let bytes = bytes.into_bytes();
185
186        let mut decoder = Decoder::new(&bytes);
187        let response = OffsetFetchResponseV2::decode_body(&mut decoder).unwrap();
188
189        assert_eq!(
190            response,
191            OffsetFetchResponseV2 {
192                topics: vec![OffsetFetchTopicResponse {
193                    name: "orders".to_owned(),
194                    partitions: vec![OffsetFetchPartitionResponse {
195                        partition_index: 0,
196                        committed_offset: 42,
197                        metadata: Some("processed".to_owned()),
198                        error_code: 0,
199                    }],
200                }],
201                error_code: 0,
202            }
203        );
204        assert!(decoder.is_empty());
205    }
206}
kafrust_protocol/api/offset_fetch.rs

kafrust_protocol/api/
offset_fetch.rs
