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        Ok(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    }
87}
88
89#[derive(Debug, Clone, PartialEq, Eq)]
90pub struct OffsetForLeaderEpochTopicResponseV3 {
91    pub name: String,
92    pub partitions: Vec<OffsetForLeaderEpochPartitionResponseV3>,
93}
94
95#[derive(Debug, Clone, PartialEq, Eq)]
96pub struct OffsetForLeaderEpochPartitionResponseV3 {
97    pub error_code: i16,
98    pub partition_index: i32,
99    pub leader_epoch: i32,
100    pub end_offset: i64,
101}
102
103#[cfg(test)]
104#[allow(clippy::unwrap_used)]
105mod tests {
106    use super::*;
107
108    #[test]
109    fn encodes_offset_for_leader_epoch_v3_request() {
110        let request = OffsetForLeaderEpochRequestV3 {
111            correlation_id: 7,
112            client_id: None,
113            replica_id: -1,
114            topics: vec![OffsetForLeaderEpochTopicV3 {
115                name: "orders".to_owned(),
116                partitions: vec![OffsetForLeaderEpochPartitionV3 {
117                    partition_index: 2,
118                    current_leader_epoch: 8,
119                    leader_epoch: 7,
120                }],
121            }],
122        };
123
124        assert_eq!(
125            request.encode().unwrap(),
126            [
127                0, 23, 0, 3, 0, 0, 0, 7, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0, 0, 0, 1, 0, 6,
128                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,
129            ]
130        );
131    }
132
133    #[test]
134    fn decodes_offset_for_leader_epoch_v3_response() {
135        let mut encoder = Encoder::new();
136        encoder.write_i32(12);
137        encoder.write_i32(1);
138        encoder.write_string("orders").unwrap();
139        encoder.write_i32(1);
140        encoder.write_i16(0);
141        encoder.write_i32(2);
142        encoder.write_i32(8);
143        encoder.write_i64(42);
144
145        let response =
146            OffsetForLeaderEpochResponseV3::decode_body(&mut Decoder::new(&encoder.into_bytes()))
147                .unwrap();
148        assert_eq!(response.throttle_time_ms, 12);
149        assert_eq!(response.topics[0].partitions[0].leader_epoch, 8);
150        assert_eq!(response.topics[0].partitions[0].end_offset, 42);
151    }
152}
kafrust_protocol/api/offset_for_leader_epoch.rs

kafrust_protocol/api/
offset_for_leader_epoch.rs
