kafrust_protocol/api/
consumer_group_heartbeat.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5/// Kafka ConsumerGroupHeartbeat API key.
6pub const API_KEY: i16 = 68;
7
8/// Topic partitions owned by a member in the KIP-848 consumer group protocol.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct ConsumerGroupHeartbeatTopicPartitions {
11    pub topic_id: [u8; 16],
12    pub partitions: Vec<i32>,
13}
14
15impl ConsumerGroupHeartbeatTopicPartitions {
16    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
17        encoder.write_uuid(&self.topic_id);
18        encoder.write_compact_array(Some(&self.partitions), |encoder, partition| {
19            encoder.write_i32(*partition);
20            Ok(())
21        })?;
22        encoder.write_empty_tagged_fields();
23        Ok(())
24    }
25
26    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
27        let topic_id = decoder.read_uuid()?;
28        let partitions = decoder
29            .read_compact_array("consumer group heartbeat partitions", |decoder| {
30                decoder.read_i32()
31            })?
32            .unwrap_or_default();
33        decoder.read_tagged_fields()?;
34        Ok(Self {
35            topic_id,
36            partitions,
37        })
38    }
39}
40
41/// KIP-848 ConsumerGroupHeartbeat v0 request.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct ConsumerGroupHeartbeatRequestV0 {
44    pub correlation_id: i32,
45    pub client_id: Option<String>,
46    pub group_id: String,
47    pub member_id: String,
48    pub member_epoch: i32,
49    pub instance_id: Option<String>,
50    pub rack_id: Option<String>,
51    pub rebalance_timeout_ms: i32,
52    pub subscribed_topic_names: Option<Vec<String>>,
53    pub server_assignor: Option<String>,
54    pub topic_partitions: Option<Vec<ConsumerGroupHeartbeatTopicPartitions>>,
55}
56
57impl ConsumerGroupHeartbeatRequestV0 {
58    pub fn encode(&self) -> Result<Vec<u8>> {
59        let mut encoder = Encoder::new();
60        RequestHeader {
61            api_key: API_KEY,
62            api_version: 0,
63            correlation_id: self.correlation_id,
64            client_id: self.client_id.clone(),
65        }
66        .encode_v2(&mut encoder)?;
67        encoder.write_compact_string(&self.group_id)?;
68        encoder.write_compact_string(&self.member_id)?;
69        encoder.write_i32(self.member_epoch);
70        encoder.write_compact_nullable_string(self.instance_id.as_deref())?;
71        encoder.write_compact_nullable_string(self.rack_id.as_deref())?;
72        encoder.write_i32(self.rebalance_timeout_ms);
73        encoder.write_compact_array(self.subscribed_topic_names.as_deref(), |encoder, topic| {
74            encoder.write_compact_string(topic)
75        })?;
76        encoder.write_compact_nullable_string(self.server_assignor.as_deref())?;
77        encoder.write_compact_array(self.topic_partitions.as_deref(), |encoder, topic| {
78            topic.encode(encoder)
79        })?;
80        encoder.write_empty_tagged_fields();
81        Ok(encoder.into_bytes())
82    }
83}
84
85/// KIP-848 ConsumerGroupHeartbeat v0 response.
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct ConsumerGroupHeartbeatResponseV0 {
88    pub throttle_time_ms: i32,
89    pub error_code: i16,
90    pub error_message: Option<String>,
91    pub member_id: Option<String>,
92    pub member_epoch: i32,
93    pub heartbeat_interval_ms: i32,
94    pub assignment: Option<Vec<ConsumerGroupHeartbeatTopicPartitions>>,
95}
96
97impl ConsumerGroupHeartbeatResponseV0 {
98    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
99        let throttle_time_ms = decoder.read_i32()?;
100        let error_code = decoder.read_i16()?;
101        let error_message = decoder.read_compact_nullable_string()?;
102        let member_id = decoder.read_compact_nullable_string()?;
103        let member_epoch = decoder.read_i32()?;
104        let heartbeat_interval_ms = decoder.read_i32()?;
105        // Assignment is a nullable struct, not a compact nullable array. The
106        // struct contains the compact topic-partitions array and its tags.
107        let assignment = match decoder.read_i8()? {
108            -1 => None,
109            1 => {
110                let topic_partitions = decoder
111                    .read_compact_array("consumer group heartbeat assignment", |decoder| {
112                        ConsumerGroupHeartbeatTopicPartitions::decode(decoder)
113                    })?
114                    .unwrap_or_default();
115                decoder.read_tagged_fields()?;
116                Some(topic_partitions)
117            }
118            marker => return Err(crate::error::Error::InvalidNullableStruct(marker)),
119        };
120        decoder.read_tagged_fields()?;
121        Ok(Self {
122            throttle_time_ms,
123            error_code,
124            error_message,
125            member_id,
126            member_epoch,
127            heartbeat_interval_ms,
128            assignment,
129        })
130    }
131}
132
133#[cfg(test)]
134#[allow(clippy::unwrap_used)]
135mod tests {
136    use super::{
137        ConsumerGroupHeartbeatRequestV0, ConsumerGroupHeartbeatResponseV0,
138        ConsumerGroupHeartbeatTopicPartitions,
139    };
140    use crate::codec::{Decoder, Encoder};
141
142    #[test]
143    fn encodes_kip_848_heartbeat_request_with_subscription_and_assignment() {
144        let request = ConsumerGroupHeartbeatRequestV0 {
145            correlation_id: 23,
146            client_id: Some("kafrust".to_owned()),
147            group_id: "orders-group".to_owned(),
148            member_id: "member-a".to_owned(),
149            member_epoch: 4,
150            instance_id: Some("instance-a".to_owned()),
151            rack_id: None,
152            rebalance_timeout_ms: 30_000,
153            subscribed_topic_names: Some(vec!["orders".to_owned()]),
154            server_assignor: Some("uniform".to_owned()),
155            topic_partitions: Some(vec![ConsumerGroupHeartbeatTopicPartitions {
156                topic_id: [1; 16],
157                partitions: vec![0, 2],
158            }]),
159        };
160
161        let encoded = request.encode().unwrap();
162        assert_eq!(&encoded[0..4], &[0, 68, 0, 0]);
163        assert_eq!(&encoded[4..12], &[0, 0, 0, 23, 0, 7, b'k', b'a']);
164        assert!(encoded.windows(12).any(|bytes| bytes == b"orders-group"));
165        assert!(encoded.windows(16).any(|bytes| bytes == [1; 16]));
166        assert_eq!(*encoded.last().unwrap(), 0);
167    }
168
169    #[test]
170    fn decodes_kip_848_heartbeat_response_and_assignment() {
171        let mut bytes = Encoder::new();
172        bytes.write_i32(12);
173        bytes.write_i16(0);
174        bytes.write_compact_nullable_string(None).unwrap();
175        bytes
176            .write_compact_nullable_string(Some("member-a"))
177            .unwrap();
178        bytes.write_i32(5);
179        bytes.write_i32(2500);
180        bytes.write_i8(1);
181        bytes
182            .write_compact_array(
183                Some(&[ConsumerGroupHeartbeatTopicPartitions {
184                    topic_id: [2; 16],
185                    partitions: vec![1, 3],
186                }]),
187                |encoder, topic| topic.encode(encoder),
188            )
189            .unwrap();
190        bytes.write_empty_tagged_fields();
191        bytes.write_empty_tagged_fields();
192
193        let bytes = bytes.into_bytes();
194        let mut decoder = Decoder::new(&bytes);
195        let response = ConsumerGroupHeartbeatResponseV0::decode_body(&mut decoder).unwrap();
196
197        assert_eq!(response.throttle_time_ms, 12);
198        assert_eq!(response.error_code, 0);
199        assert_eq!(response.member_id.as_deref(), Some("member-a"));
200        assert_eq!(response.member_epoch, 5);
201        assert_eq!(response.heartbeat_interval_ms, 2500);
202        assert_eq!(response.assignment.unwrap()[0].partitions, vec![1, 3]);
203        assert!(decoder.is_empty());
204    }
205
206    #[test]
207    fn decodes_kip_848_heartbeat_response_with_null_assignment_struct() {
208        let mut bytes = Encoder::new();
209        bytes.write_i32(0);
210        bytes.write_i16(0);
211        bytes.write_compact_nullable_string(None).unwrap();
212        bytes
213            .write_compact_nullable_string(Some("member-a"))
214            .unwrap();
215        bytes.write_i32(1);
216        bytes.write_i32(2500);
217        bytes.write_i8(-1);
218        bytes.write_empty_tagged_fields();
219
220        let bytes = bytes.into_bytes();
221        let mut decoder = Decoder::new(&bytes);
222        let response = ConsumerGroupHeartbeatResponseV0::decode_body(&mut decoder).unwrap();
223
224        assert!(response.assignment.is_none());
225        assert!(decoder.is_empty());
226    }
227
228    #[test]
229    fn encodes_nullable_kip_848_fields_as_null_arrays() {
230        let request = ConsumerGroupHeartbeatRequestV0 {
231            correlation_id: 1,
232            client_id: None,
233            group_id: "g".to_owned(),
234            member_id: "m".to_owned(),
235            member_epoch: -1,
236            instance_id: None,
237            rack_id: None,
238            rebalance_timeout_ms: -1,
239            subscribed_topic_names: None,
240            server_assignor: None,
241            topic_partitions: None,
242        };
243
244        let encoded = request.encode().unwrap();
245        assert_eq!(&encoded[0..10], &[0, 68, 0, 0, 0, 0, 0, 1, 0xff, 0xff]);
246        assert_eq!(encoded[10], 0); // request-header tagged fields
247        assert!(encoded.ends_with(&[0, 0, 0]));
248    }
249}
kafrust_protocol/api/consumer_group_heartbeat.rs

kafrust_protocol/api/
consumer_group_heartbeat.rs
