kafrust_protocol/api/
join_group.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 11;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct JoinGroupRequestV2 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub group_id: String,
12    pub session_timeout_ms: i32,
13    pub rebalance_timeout_ms: i32,
14    pub member_id: String,
15    pub protocol_type: String,
16    pub protocols: Vec<JoinGroupProtocol>,
17}
18
19impl JoinGroupRequestV2 {
20    pub fn encode(&self) -> Result<Vec<u8>> {
21        let mut encoder = Encoder::new();
22        RequestHeader {
23            api_key: API_KEY,
24            api_version: 2,
25            correlation_id: self.correlation_id,
26            client_id: self.client_id.clone(),
27        }
28        .encode_v1(&mut encoder)?;
29        encoder.write_string(&self.group_id)?;
30        encoder.write_i32(self.session_timeout_ms);
31        encoder.write_i32(self.rebalance_timeout_ms);
32        encoder.write_string(&self.member_id)?;
33        encoder.write_string(&self.protocol_type)?;
34        encoder.write_array(Some(self.protocols.as_slice()), |encoder, protocol| {
35            protocol.encode(encoder)
36        })?;
37        Ok(encoder.into_bytes())
38    }
39}
40
41#[derive(Debug, Clone, PartialEq, Eq)]
42pub struct JoinGroupRequestV5 {
43    pub correlation_id: i32,
44    pub client_id: Option<String>,
45    pub group_id: String,
46    pub session_timeout_ms: i32,
47    pub rebalance_timeout_ms: i32,
48    pub member_id: String,
49    pub group_instance_id: Option<String>,
50    pub protocol_type: String,
51    pub protocols: Vec<JoinGroupProtocol>,
52}
53
54impl JoinGroupRequestV5 {
55    pub fn encode(&self) -> Result<Vec<u8>> {
56        let mut encoder = Encoder::new();
57        RequestHeader {
58            api_key: API_KEY,
59            api_version: 5,
60            correlation_id: self.correlation_id,
61            client_id: self.client_id.clone(),
62        }
63        .encode_v1(&mut encoder)?;
64        encoder.write_string(&self.group_id)?;
65        encoder.write_i32(self.session_timeout_ms);
66        encoder.write_i32(self.rebalance_timeout_ms);
67        encoder.write_string(&self.member_id)?;
68        encoder.write_nullable_string(self.group_instance_id.as_deref())?;
69        encoder.write_string(&self.protocol_type)?;
70        encoder.write_array(Some(self.protocols.as_slice()), |encoder, protocol| {
71            protocol.encode(encoder)
72        })?;
73        Ok(encoder.into_bytes())
74    }
75}
76
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub struct JoinGroupProtocol {
79    pub name: String,
80    pub metadata: Vec<u8>,
81}
82
83impl JoinGroupProtocol {
84    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
85        encoder.write_string(&self.name)?;
86        encoder.write_bytes(&self.metadata)
87    }
88}
89
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub struct JoinGroupResponseV2 {
92    pub throttle_time_ms: i32,
93    pub error_code: i16,
94    pub generation_id: i32,
95    pub protocol_name: String,
96    pub leader: String,
97    pub member_id: String,
98    pub members: Vec<JoinGroupMember>,
99}
100
101impl JoinGroupResponseV2 {
102    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
103        Ok(Self {
104            throttle_time_ms: decoder.read_i32()?,
105            error_code: decoder.read_i16()?,
106            generation_id: decoder.read_i32()?,
107            protocol_name: decoder.read_string()?,
108            leader: decoder.read_string()?,
109            member_id: decoder.read_string()?,
110            members: decoder
111                .read_array("join group members", JoinGroupMember::decode)?
112                .unwrap_or_default(),
113        })
114    }
115}
116
117#[derive(Debug, Clone, PartialEq, Eq)]
118pub struct JoinGroupResponseV5 {
119    pub throttle_time_ms: i32,
120    pub error_code: i16,
121    pub generation_id: i32,
122    pub protocol_name: String,
123    pub leader: String,
124    pub member_id: String,
125    pub members: Vec<JoinGroupMemberV5>,
126}
127
128impl JoinGroupResponseV5 {
129    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
130        Ok(Self {
131            throttle_time_ms: decoder.read_i32()?,
132            error_code: decoder.read_i16()?,
133            generation_id: decoder.read_i32()?,
134            protocol_name: decoder.read_string()?,
135            leader: decoder.read_string()?,
136            member_id: decoder.read_string()?,
137            members: decoder
138                .read_array("join group members", JoinGroupMemberV5::decode)?
139                .unwrap_or_default(),
140        })
141    }
142}
143
144#[derive(Debug, Clone, PartialEq, Eq)]
145pub struct JoinGroupMember {
146    pub member_id: String,
147    pub metadata: Vec<u8>,
148}
149
150impl JoinGroupMember {
151    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
152        Ok(Self {
153            member_id: decoder.read_string()?,
154            metadata: decoder.read_bytes()?,
155        })
156    }
157}
158
159#[derive(Debug, Clone, PartialEq, Eq)]
160pub struct JoinGroupMemberV5 {
161    pub member_id: String,
162    pub group_instance_id: Option<String>,
163    pub metadata: Vec<u8>,
164}
165
166impl JoinGroupMemberV5 {
167    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
168        Ok(Self {
169            member_id: decoder.read_string()?,
170            group_instance_id: decoder.read_nullable_string()?,
171            metadata: decoder.read_bytes()?,
172        })
173    }
174}
175
176#[cfg(test)]
177#[allow(clippy::unwrap_used)]
178mod tests {
179    use super::{
180        JoinGroupMember, JoinGroupMemberV5, JoinGroupProtocol, JoinGroupRequestV2,
181        JoinGroupRequestV5, JoinGroupResponseV2, JoinGroupResponseV5,
182    };
183    use crate::codec::{Decoder, Encoder};
184
185    #[test]
186    fn encodes_join_group_v2_request() {
187        let request = JoinGroupRequestV2 {
188            correlation_id: 13,
189            client_id: Some("kafrust".to_owned()),
190            group_id: "orders-group".to_owned(),
191            session_timeout_ms: 10_000,
192            rebalance_timeout_ms: 30_000,
193            member_id: String::new(),
194            protocol_type: "consumer".to_owned(),
195            protocols: vec![JoinGroupProtocol {
196                name: "range".to_owned(),
197                metadata: vec![1, 2, 3],
198            }],
199        };
200
201        assert_eq!(
202            request.encode().unwrap(),
203            [
204                0, 11, // api key
205                0, 2, // api version
206                0, 0, 0, 13, // correlation id
207                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
208                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
209                b'p', // group id
210                0, 0, 39, 16, // session timeout
211                0, 0, 117, 48, // rebalance timeout
212                0, 0, // member id
213                0, 8, b'c', b'o', b'n', b's', b'u', b'm', b'e', b'r', // protocol type
214                0, 0, 0, 1, // protocol count
215                0, 5, b'r', b'a', b'n', b'g', b'e', // protocol name
216                0, 0, 0, 3, 1, 2, 3, // metadata
217            ]
218        );
219    }
220
221    #[test]
222    fn decodes_join_group_v2_response() {
223        let mut bytes = Encoder::new();
224        bytes.write_i32(0);
225        bytes.write_i16(0);
226        bytes.write_i32(7);
227        bytes.write_string("range").unwrap();
228        bytes.write_string("member-a").unwrap();
229        bytes.write_string("member-a").unwrap();
230        bytes.write_i32(1);
231        bytes.write_string("member-a").unwrap();
232        bytes.write_bytes(&[1, 2, 3]).unwrap();
233        let bytes = bytes.into_bytes();
234
235        let mut decoder = Decoder::new(&bytes);
236        let response = JoinGroupResponseV2::decode_body(&mut decoder).unwrap();
237
238        assert_eq!(response.throttle_time_ms, 0);
239        assert_eq!(response.error_code, 0);
240        assert_eq!(response.generation_id, 7);
241        assert_eq!(response.protocol_name, "range");
242        assert_eq!(response.leader, "member-a");
243        assert_eq!(response.member_id, "member-a");
244        assert_eq!(
245            response.members,
246            vec![JoinGroupMember {
247                member_id: "member-a".to_owned(),
248                metadata: vec![1, 2, 3],
249            }]
250        );
251        assert!(decoder.is_empty());
252    }
253
254    #[test]
255    fn encodes_join_group_v5_request_with_static_member() {
256        let request = JoinGroupRequestV5 {
257            correlation_id: 13,
258            client_id: Some("kafrust".to_owned()),
259            group_id: "orders-group".to_owned(),
260            session_timeout_ms: 10_000,
261            rebalance_timeout_ms: 30_000,
262            member_id: "member-a".to_owned(),
263            group_instance_id: Some("orders-reader-1".to_owned()),
264            protocol_type: "consumer".to_owned(),
265            protocols: vec![JoinGroupProtocol {
266                name: "range".to_owned(),
267                metadata: vec![1, 2, 3],
268            }],
269        };
270
271        let encoded = request.encode().unwrap();
272        assert_eq!(&encoded[0..4], &[0, 11, 0, 5]);
273        assert!(encoded
274            .windows(17)
275            .any(|bytes| bytes == b"\0\x0forders-reader-1"));
276    }
277
278    #[test]
279    fn decodes_join_group_v5_response_with_static_member() {
280        let mut bytes = Encoder::new();
281        bytes.write_i32(0);
282        bytes.write_i16(0);
283        bytes.write_i32(7);
284        bytes.write_string("range").unwrap();
285        bytes.write_string("member-a").unwrap();
286        bytes.write_string("member-a").unwrap();
287        bytes.write_i32(1);
288        bytes.write_string("member-a").unwrap();
289        bytes
290            .write_nullable_string(Some("orders-reader-1"))
291            .unwrap();
292        bytes.write_bytes(&[1, 2, 3]).unwrap();
293        let bytes = bytes.into_bytes();
294
295        let mut decoder = Decoder::new(&bytes);
296        let response = JoinGroupResponseV5::decode_body(&mut decoder).unwrap();
297
298        assert_eq!(
299            response.members,
300            vec![JoinGroupMemberV5 {
301                member_id: "member-a".to_owned(),
302                group_instance_id: Some("orders-reader-1".to_owned()),
303                metadata: vec![1, 2, 3],
304            }]
305        );
306        assert!(decoder.is_empty());
307    }
308}
kafrust_protocol/api/join_group.rs

kafrust_protocol/api/
join_group.rs
