kafrust_protocol/api/
offset_for_leader_epoch.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5/// Kafka OffsetForLeaderEpoch API key.
6pub const API_KEY: i16 = 23;
7
8/// OffsetForLeaderEpoch v3 request.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct OffsetForLeaderEpochRequestV3 {
11    pub correlation_id: i32,
12    pub client_id: Option<String>,
13    pub replica_id: i32,
14    pub topics: Vec<OffsetForLeaderEpochTopicV3>,
15}
16
17impl OffsetForLeaderEpochRequestV3 {
18    pub fn encode(&self) -> Result<Vec<u8>> {
19        let mut encoder = Encoder::new();
20        RequestHeader {
21            api_key: API_KEY,
22            api_version: 3,
23            correlation_id: self.correlation_id,
24            client_id: self.client_id.clone(),
25        }
26        .encode_v1(&mut encoder)?;
27        encoder.write_i32(self.replica_id);
28        encoder.write_array(Some(&self.topics), |encoder, topic| topic.encode(encoder))?;
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct OffsetForLeaderEpochTopicV3 {
35    pub name: String,
36    pub partitions: Vec<OffsetForLeaderEpochPartitionV3>,
37}
38
39impl OffsetForLeaderEpochTopicV3 {
40    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
41        encoder.write_string(&self.name)?;
42        encoder.write_array(Some(&self.partitions), |encoder, partition| {
43            encoder.write_i32(partition.partition_index);
44            encoder.write_i32(partition.current_leader_epoch);
45            encoder.write_i32(partition.leader_epoch);
46            Ok(())
47        })
48    }
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct OffsetForLeaderEpochPartitionV3 {
53    pub partition_index: i32,
54    pub current_leader_epoch: i32,
55    pub leader_epoch: i32,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct OffsetForLeaderEpochResponseV3 {
60    pub throttle_time_ms: i32,
61    pub topics: Vec<OffsetForLeaderEpochTopicResponseV3>,
62}
63
64impl OffsetForLeaderEpochResponseV3 {
65    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
66        let response = Self {
67            throttle_time_ms: decoder.read_i32()?,
68            topics: decoder
69                .read_array("offset for leader epoch topic responses", |decoder| {
70                    Ok(OffsetForLeaderEpochTopicResponseV3 {
71                        name: decoder.read_string()?,
72                        partitions: decoder
73                            .read_array("offset for leader epoch partition responses", |decoder| {
74                                Ok(OffsetForLeaderEpochPartitionResponseV3 {
75                                    error_code: decoder.read_i16()?,
76                                    partition_index: decoder.read_i32()?,
77                                    leader_epoch: decoder.read_i32()?,
78                                    end_offset: decoder.read_i64()?,
79                                })
80                            })?
81                            .unwrap_or_default(),
82                    })
83                })?
84                .unwrap_or_default(),
85        };
86        decoder.finish()?;
87        Ok(response)
88    }
89}
90
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub struct OffsetForLeaderEpochTopicResponseV3 {
93    pub name: String,
94    pub partitions: Vec<OffsetForLeaderEpochPartitionResponseV3>,
95}
96
97#[derive(Debug, Clone, PartialEq, Eq)]
98pub struct OffsetForLeaderEpochPartitionResponseV3 {
99    pub error_code: i16,
100    pub partition_index: i32,
101    pub leader_epoch: i32,
102    pub end_offset: i64,
103}
104
105#[cfg(test)]
106#[allow(clippy::unwrap_used)]
107mod tests {
108    use super::*;
109
110    #[test]
111    fn encodes_offset_for_leader_epoch_v3_request() {
112        let request = OffsetForLeaderEpochRequestV3 {
113            correlation_id: 7,
114            client_id: None,
115            replica_id: -1,
116            topics: vec![OffsetForLeaderEpochTopicV3 {
117                name: "orders".to_owned(),
118                partitions: vec![OffsetForLeaderEpochPartitionV3 {
119                    partition_index: 2,
120                    current_leader_epoch: 8,
121                    leader_epoch: 7,
122                }],
123            }],
124        };
125
126        assert_eq!(
127            request.encode().unwrap(),
128            [
129                0, 23, 0, 3, 0, 0, 0, 7, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0, 0, 0, 1, 0, 6,
130                b'o', b'r', b'd', b'e', b'r', b's', 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 8, 0, 0, 0, 7,
131            ]
132        );
133    }
134
135    #[test]
136    fn decodes_offset_for_leader_epoch_v3_response() {
137        let mut encoder = Encoder::new();
138        encoder.write_i32(12);
139        encoder.write_i32(1);
140        encoder.write_string("orders").unwrap();
141        encoder.write_i32(1);
142        encoder.write_i16(0);
143        encoder.write_i32(2);
144        encoder.write_i32(8);
145        encoder.write_i64(42);
146
147        let response =
148            OffsetForLeaderEpochResponseV3::decode_body(&mut Decoder::new(&encoder.into_bytes()))
149                .unwrap();
150        assert_eq!(response.throttle_time_ms, 12);
151        assert_eq!(response.topics[0].partitions[0].leader_epoch, 8);
152        assert_eq!(response.topics[0].partitions[0].end_offset, 42);
153    }
154}
kafrust_protocol/api/offset_for_leader_epoch.rs

kafrust_protocol/api/
offset_for_leader_epoch.rs
