kafrust-protocol 0.2.19

Kafka wire protocol primitives for kafrust.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

use crate::codec::{Decoder, Encoder};
use crate::error::Result;
use crate::header::RequestHeader;

/// Kafka OffsetForLeaderEpoch API key.
pub const API_KEY: i16 = 23;

/// OffsetForLeaderEpoch v3 request.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OffsetForLeaderEpochRequestV3 {
    pub correlation_id: i32,
    pub client_id: Option<String>,
    pub replica_id: i32,
    pub topics: Vec<OffsetForLeaderEpochTopicV3>,
}

impl OffsetForLeaderEpochRequestV3 {
    pub fn encode(&self) -> Result<Vec<u8>> {
        let mut encoder = Encoder::new();
        RequestHeader {
            api_key: API_KEY,
            api_version: 3,
            correlation_id: self.correlation_id,
            client_id: self.client_id.clone(),
        }
        .encode_v1(&mut encoder)?;
        encoder.write_i32(self.replica_id);
        encoder.write_array(Some(&self.topics), |encoder, topic| topic.encode(encoder))?;
        Ok(encoder.into_bytes())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OffsetForLeaderEpochTopicV3 {
    pub name: String,
    pub partitions: Vec<OffsetForLeaderEpochPartitionV3>,
}

impl OffsetForLeaderEpochTopicV3 {
    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
        encoder.write_string(&self.name)?;
        encoder.write_array(Some(&self.partitions), |encoder, partition| {
            encoder.write_i32(partition.partition_index);
            encoder.write_i32(partition.current_leader_epoch);
            encoder.write_i32(partition.leader_epoch);
            Ok(())
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OffsetForLeaderEpochPartitionV3 {
    pub partition_index: i32,
    pub current_leader_epoch: i32,
    pub leader_epoch: i32,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OffsetForLeaderEpochResponseV3 {
    pub throttle_time_ms: i32,
    pub topics: Vec<OffsetForLeaderEpochTopicResponseV3>,
}

impl OffsetForLeaderEpochResponseV3 {
    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
        Ok(Self {
            throttle_time_ms: decoder.read_i32()?,
            topics: decoder
                .read_array("offset for leader epoch topic responses", |decoder| {
                    Ok(OffsetForLeaderEpochTopicResponseV3 {
                        name: decoder.read_string()?,
                        partitions: decoder
                            .read_array("offset for leader epoch partition responses", |decoder| {
                                Ok(OffsetForLeaderEpochPartitionResponseV3 {
                                    error_code: decoder.read_i16()?,
                                    partition_index: decoder.read_i32()?,
                                    leader_epoch: decoder.read_i32()?,
                                    end_offset: decoder.read_i64()?,
                                })
                            })?
                            .unwrap_or_default(),
                    })
                })?
                .unwrap_or_default(),
        })
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OffsetForLeaderEpochTopicResponseV3 {
    pub name: String,
    pub partitions: Vec<OffsetForLeaderEpochPartitionResponseV3>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OffsetForLeaderEpochPartitionResponseV3 {
    pub error_code: i16,
    pub partition_index: i32,
    pub leader_epoch: i32,
    pub end_offset: i64,
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;

    #[test]
    fn encodes_offset_for_leader_epoch_v3_request() {
        let request = OffsetForLeaderEpochRequestV3 {
            correlation_id: 7,
            client_id: None,
            replica_id: -1,
            topics: vec![OffsetForLeaderEpochTopicV3 {
                name: "orders".to_owned(),
                partitions: vec![OffsetForLeaderEpochPartitionV3 {
                    partition_index: 2,
                    current_leader_epoch: 8,
                    leader_epoch: 7,
                }],
            }],
        };

        assert_eq!(
            request.encode().unwrap(),
            [
                0, 23, 0, 3, 0, 0, 0, 7, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0, 0, 0, 1, 0, 6,
                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,
            ]
        );
    }

    #[test]
    fn decodes_offset_for_leader_epoch_v3_response() {
        let mut encoder = Encoder::new();
        encoder.write_i32(12);
        encoder.write_i32(1);
        encoder.write_string("orders").unwrap();
        encoder.write_i32(1);
        encoder.write_i16(0);
        encoder.write_i32(2);
        encoder.write_i32(8);
        encoder.write_i64(42);

        let response =
            OffsetForLeaderEpochResponseV3::decode_body(&mut Decoder::new(&encoder.into_bytes()))
                .unwrap();
        assert_eq!(response.throttle_time_ms, 12);
        assert_eq!(response.topics[0].partitions[0].leader_epoch, 8);
        assert_eq!(response.topics[0].partitions[0].end_offset, 42);
    }
}