krabka_protocol/records/metadata/
envelope.rs

1//! The `KRaft` metadata record-value envelope (`MetadataRecordSerde` /
2//! `ApiMessageAndVersion`): a record value is
3//! `frameVersion (uvarint, 1) + apiKey (uvarint) + apiVersion (uvarint) + body`.
4
5use bytes::{Buf, BufMut, Bytes, BytesMut};
6
7use crate::primitives::varint::{get_uvarint, put_uvarint, uvarint_len};
8
9/// The only `KRaft` metadata frame version, which is part of the krabka 1.x
10/// on-disk contract. Kafka's `AbstractApiMessageSerde.DEFAULT_FRAME_VERSION`
11/// is 1, and the `__cluster_metadata` record values of
12/// mirror.gcr.io/apache/kafka:4.0.0 start with `0x01`.
13pub const FRAME_VERSION: u32 = 1;
14
15/// Decoded envelope header, which is everything before the message body.
16///
17/// It has no frame version: [`decode_value_header`] accepts only
18/// [`FRAME_VERSION`].
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub struct ValueHeader {
21    pub api_key: u32,
22    pub api_version: u32,
23}
24
25/// Error decoding a metadata record value envelope.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
27pub enum EnvelopeError {
28    #[error("truncated metadata record value envelope")]
29    Truncated,
30    /// The envelope declares a frame version other than [`FRAME_VERSION`].
31    /// The text matches the `MetadataParseException` of Kafka's
32    /// `AbstractApiMessageSerde.read`.
33    #[error(
34        "Could not deserialize metadata record due to unknown frame version {0}(only frame version 1 is supported)"
35    )]
36    UnknownFrameVersion(u32),
37}
38
39/// Encodes a record value: the envelope header plus the already-encoded `body`
40/// bytes.
41#[must_use]
42pub fn encode_value(api_key: u32, api_version: u32, body: &[u8]) -> Bytes {
43    let mut out = BytesMut::with_capacity(
44        uvarint_len(FRAME_VERSION) + uvarint_len(api_key) + uvarint_len(api_version) + body.len(),
45    );
46    put_uvarint(&mut out, FRAME_VERSION);
47    put_uvarint(&mut out, api_key);
48    put_uvarint(&mut out, api_version);
49    out.put_slice(body);
50    out.freeze()
51}
52
53/// Decodes the envelope header and leaves `buf` positioned at the message body.
54///
55/// # Errors
56/// Returns [`EnvelopeError::Truncated`] if any varint cannot be read, and
57/// [`EnvelopeError::UnknownFrameVersion`] for a frame version other than
58/// [`FRAME_VERSION`], as Kafka's `AbstractApiMessageSerde.read` refuses one.
59pub fn decode_value_header<B: Buf>(buf: &mut B) -> Result<ValueHeader, EnvelopeError> {
60    let frame_version = get_uvarint(buf).map_err(|_| EnvelopeError::Truncated)?;
61    if frame_version != FRAME_VERSION {
62        return Err(EnvelopeError::UnknownFrameVersion(frame_version));
63    }
64    let api_key = get_uvarint(buf).map_err(|_| EnvelopeError::Truncated)?;
65    let api_version = get_uvarint(buf).map_err(|_| EnvelopeError::Truncated)?;
66    Ok(ValueHeader {
67        api_key,
68        api_version,
69    })
70}
71
72#[cfg(test)]
73mod tests {
74
75    use super::*;
76
77    /// `FeatureLevelRecord` apiKey 12, apiVersion 0, with body `01 02`, as
78    /// Kafka frames it. These bytes are the 1.x on-disk contract.
79    const GOLDEN_VALUE: [u8; 5] = [0x01, 0x0C, 0x00, 0x01, 0x02];
80
81    #[test]
82    fn encode_matches_the_golden_bytes() {
83        assert2::assert!(encode_value(12, 0, &[0x01, 0x02])[..] == GOLDEN_VALUE);
84    }
85
86    #[test]
87    fn golden_bytes_decode_to_the_header_and_body() {
88        let mut cur: &[u8] = &GOLDEN_VALUE;
89        let hdr = decode_value_header(&mut cur);
90        assert2::assert!(
91            hdr == Ok(ValueHeader {
92                api_key: 12,
93                api_version: 0,
94            })
95        );
96        assert2::assert!(cur == [0x01, 0x02]);
97    }
98
99    #[test]
100    fn rejects_a_bad_envelope() {
101        for (label, value, want) in [
102            ("empty", vec![], EnvelopeError::Truncated),
103            ("frame version only", vec![0x01], EnvelopeError::Truncated),
104            (
105                "frame version 0",
106                vec![0x00, 0x0C, 0x00],
107                EnvelopeError::UnknownFrameVersion(0),
108            ),
109            (
110                "frame version 2",
111                vec![0x02, 0x0C, 0x00],
112                EnvelopeError::UnknownFrameVersion(2),
113            ),
114            (
115                "frame version 300 (two-byte varint)",
116                vec![0xAC, 0x02, 0x0C, 0x00],
117                EnvelopeError::UnknownFrameVersion(300),
118            ),
119        ] {
120            let mut cur: &[u8] = &value;
121            assert2::assert!(decode_value_header(&mut cur) == Err(want), "{label}");
122        }
123    }
124
125    #[test]
126    fn unknown_frame_version_reads_as_kafkas_message() {
127        assert2::assert!(
128            EnvelopeError::UnknownFrameVersion(2).to_string()
129                == "Could not deserialize metadata record due to unknown frame version 2(only frame version 1 is supported)"
130        );
131    }
132}
krabka_protocol/records/metadata/envelope.rs

krabka_protocol/records/metadata/
envelope.rs
