kafrust_protocol/api/
find_coordinator.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 10;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub enum CoordinatorType {
9    Group,
10    Transaction,
11    Share,
12}
13
14impl CoordinatorType {
15    fn as_i8(self) -> i8 {
16        match self {
17            Self::Group => 0,
18            Self::Transaction => 1,
19            Self::Share => 2,
20        }
21    }
22}
23
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub struct FindCoordinatorRequestV1 {
26    pub correlation_id: i32,
27    pub client_id: Option<String>,
28    pub coordinator_key: String,
29    pub coordinator_type: CoordinatorType,
30}
31
32impl FindCoordinatorRequestV1 {
33    pub fn encode(&self) -> Result<Vec<u8>> {
34        let mut encoder = Encoder::new();
35        RequestHeader {
36            api_key: API_KEY,
37            api_version: 1,
38            correlation_id: self.correlation_id,
39            client_id: self.client_id.clone(),
40        }
41        .encode_v1(&mut encoder)?;
42        encoder.write_string(&self.coordinator_key)?;
43        encoder.write_i8(self.coordinator_type.as_i8());
44        Ok(encoder.into_bytes())
45    }
46}
47
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct FindCoordinatorResponseV1 {
50    pub throttle_time_ms: i32,
51    pub error_code: i16,
52    pub error_message: Option<String>,
53    pub node_id: i32,
54    pub host: String,
55    pub port: i32,
56}
57
58/// FindCoordinator v6 request for share-partition coordinators.
59///
60/// Kafka 4.x uses this flexible request for the KIP-932 share coordinator
61/// lookup. Each key is a `group:topic-id:partition` resource identifier.
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct FindCoordinatorRequestV6 {
64    pub correlation_id: i32,
65    pub client_id: Option<String>,
66    pub coordinator_type: CoordinatorType,
67    pub coordinator_keys: Vec<String>,
68}
69
70impl FindCoordinatorRequestV6 {
71    pub fn encode(&self) -> Result<Vec<u8>> {
72        let mut encoder = Encoder::new();
73        RequestHeader {
74            api_key: API_KEY,
75            api_version: 6,
76            correlation_id: self.correlation_id,
77            client_id: self.client_id.clone(),
78        }
79        .encode_v2(&mut encoder)?;
80        encoder.write_i8(self.coordinator_type.as_i8());
81        encoder.write_compact_array(Some(&self.coordinator_keys), |encoder, key| {
82            encoder.write_compact_string(key)
83        })?;
84        encoder.write_empty_tagged_fields();
85        Ok(encoder.into_bytes())
86    }
87}
88
89/// One coordinator result returned by FindCoordinator v6.
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub struct FindCoordinatorResultV6 {
92    pub coordinator_key: String,
93    pub node_id: i32,
94    pub host: String,
95    pub port: i32,
96    pub error_code: i16,
97    pub error_message: Option<String>,
98}
99
100/// FindCoordinator v6 response for one or more coordinator keys.
101#[derive(Debug, Clone, PartialEq, Eq)]
102pub struct FindCoordinatorResponseV6 {
103    pub throttle_time_ms: i32,
104    pub coordinators: Vec<FindCoordinatorResultV6>,
105}
106
107impl FindCoordinatorResponseV6 {
108    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
109        let throttle_time_ms = decoder.read_i32()?;
110        let coordinators = decoder
111            .read_compact_array("find coordinator results", |decoder| {
112                let result = FindCoordinatorResultV6 {
113                    coordinator_key: decoder.read_compact_string()?,
114                    node_id: decoder.read_i32()?,
115                    host: decoder.read_compact_string()?,
116                    port: decoder.read_i32()?,
117                    error_code: decoder.read_i16()?,
118                    error_message: decoder.read_compact_nullable_string()?,
119                };
120                decoder.read_tagged_fields()?;
121                Ok(result)
122            })?
123            .unwrap_or_default();
124        decoder.read_tagged_fields()?;
125        Ok(Self {
126            throttle_time_ms,
127            coordinators,
128        })
129    }
130}
131
132impl FindCoordinatorResponseV1 {
133    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
134        Ok(Self {
135            throttle_time_ms: decoder.read_i32()?,
136            error_code: decoder.read_i16()?,
137            error_message: decoder.read_nullable_string()?,
138            node_id: decoder.read_i32()?,
139            host: decoder.read_string()?,
140            port: decoder.read_i32()?,
141        })
142    }
143}
144
145#[cfg(test)]
146#[allow(clippy::unwrap_used)]
147mod tests {
148    use super::{
149        CoordinatorType, FindCoordinatorRequestV1, FindCoordinatorRequestV6,
150        FindCoordinatorResponseV1, FindCoordinatorResponseV6, API_KEY,
151    };
152    use crate::codec::Decoder;
153
154    #[test]
155    fn encodes_find_coordinator_v1_for_group() {
156        let request = FindCoordinatorRequestV1 {
157            correlation_id: 11,
158            client_id: Some("kafrust".to_owned()),
159            coordinator_key: "orders-group".to_owned(),
160            coordinator_type: CoordinatorType::Group,
161        };
162
163        assert_eq!(
164            request.encode().unwrap(),
165            [
166                0, 10, // api key
167                0, 1, // api version
168                0, 0, 0, 11, // correlation id
169                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
170                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
171                b'p', // coordinator key
172                0,    // group coordinator type
173            ]
174        );
175        assert_eq!(API_KEY, 10);
176    }
177
178    #[test]
179    fn encodes_find_coordinator_v1_for_transaction() {
180        let request = FindCoordinatorRequestV1 {
181            correlation_id: 12,
182            client_id: None,
183            coordinator_key: "orders-tx".to_owned(),
184            coordinator_type: CoordinatorType::Transaction,
185        };
186
187        assert_eq!(request.encode().unwrap().last(), Some(&1));
188    }
189
190    #[test]
191    fn encodes_find_coordinator_v1_for_share_group() {
192        let request = FindCoordinatorRequestV1 {
193            correlation_id: 13,
194            client_id: Some("kafrust".to_owned()),
195            coordinator_key: "share-group".to_owned(),
196            coordinator_type: CoordinatorType::Share,
197        };
198
199        assert_eq!(request.encode().unwrap().last(), Some(&2));
200    }
201
202    #[test]
203    fn encodes_find_coordinator_v6_for_share_partition() {
204        let request = FindCoordinatorRequestV6 {
205            correlation_id: 14,
206            client_id: Some("kafrust".to_owned()),
207            coordinator_type: CoordinatorType::Share,
208            coordinator_keys: vec!["share-orders:AQAAAAAAAAAAAAAAAAAAAA:0".to_owned()],
209        };
210
211        let encoded = request.encode().unwrap();
212        assert_eq!(&encoded[0..4], &[0, 10, 0, 6]);
213        assert_eq!(encoded[17], 0); // request header tagged fields
214        assert_eq!(encoded[18], 2); // Share coordinator type
215        assert_eq!(encoded.last(), Some(&0)); // request body tagged fields
216    }
217
218    #[test]
219    fn decodes_find_coordinator_v1_response() {
220        let bytes = [
221            0, 0, 0, 0, // throttle time
222            0, 0, // error code
223            0xff, 0xff, // null error message
224            0, 0, 0, 2, // node id
225            0, 9, b'l', b'o', b'c', b'a', b'l', b'h', b'o', b's', b't', // host
226            0, 0, 35, 132, // port 9092
227        ];
228        let mut decoder = Decoder::new(&bytes);
229        let response = FindCoordinatorResponseV1::decode_body(&mut decoder).unwrap();
230
231        assert_eq!(response.throttle_time_ms, 0);
232        assert_eq!(response.error_code, 0);
233        assert_eq!(response.error_message, None);
234        assert_eq!(response.node_id, 2);
235        assert_eq!(response.host, "localhost");
236        assert_eq!(response.port, 9092);
237        assert!(decoder.is_empty());
238    }
239
240    #[test]
241    fn decodes_find_coordinator_v6_response() {
242        let mut bytes = Vec::new();
243        bytes.extend_from_slice(&0_i32.to_be_bytes());
244        bytes.push(2); // one coordinator
245        bytes.push(7); // compact string length + 1
246        bytes.extend_from_slice(b"orders");
247        bytes.extend_from_slice(&3_i32.to_be_bytes());
248        bytes.push(7); // compact string length + 1
249        bytes.extend_from_slice(b"broker");
250        bytes.extend_from_slice(&9092_i32.to_be_bytes());
251        bytes.extend_from_slice(&0_i16.to_be_bytes());
252        bytes.push(0); // null compact error message
253        bytes.push(0); // coordinator tagged fields
254        bytes.push(0); // response tagged fields
255
256        let mut decoder = Decoder::new(&bytes);
257        let response = FindCoordinatorResponseV6::decode_body(&mut decoder).unwrap();
258        assert_eq!(response.throttle_time_ms, 0);
259        assert_eq!(response.coordinators.len(), 1);
260        assert_eq!(response.coordinators[0].coordinator_key, "orders");
261        assert_eq!(response.coordinators[0].node_id, 3);
262        assert_eq!(response.coordinators[0].host, "broker");
263        assert_eq!(response.coordinators[0].port, 9092);
264        assert_eq!(response.coordinators[0].error_code, 0);
265        assert_eq!(response.coordinators[0].error_message, None);
266        assert!(decoder.is_empty());
267    }
268}
kafrust_protocol/api/find_coordinator.rs

kafrust_protocol/api/
find_coordinator.rs
