kafrust_protocol/api/
offset_commit.rs

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

kafrust_protocol/api/
offset_commit.rs
