krabka_client_streams/processor/
serde.rs

1//! `Serde<T>`: typed (de)serialization at source/sink boundaries.
2
3use bytes::Bytes;
4
5/// Failure to deserialize bytes into `T`.
6#[derive(Debug, thiserror::Error)]
7#[error("deserialization error: {0}")]
8pub struct SerdeError(pub String);
9
10/// Serialize a `T` to bytes and back. Source nodes deserialize with it, and sink
11/// and repartition nodes serialize with it.
12#[diagnostic::on_unimplemented(
13    message = "the type `{Self}` is not a valid serializer/deserializer for `{T}`",
14    label = "does not implement `Serde<{T}>`",
15    note = "implement `Serde<{T}>` for `{Self}` or verify that the type parameter `{T}` matches the deserialized type of this serde"
16)]
17pub trait Serde<T>: Send + Sync + 'static {
18    fn serialize(&self, topic: &str, value: &T) -> Bytes;
19    /// # Errors
20    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
21    fn deserialize(&self, topic: &str, bytes: &[u8]) -> Result<T, SerdeError>;
22
23    /// Pre-register any per-topic state, such as a schema-registry subject,
24    /// before processing starts. The topology calls this method once per
25    /// `(topic, role)` at wiring time. The default is a no-op, because a
26    /// stateless serde needs nothing.
27    fn prepare(&self, _topic: &str, _role: SerdeRole) {}
28}
29
30/// Whether a serde is wired for the record key or the record value. The
31/// topology passes this to [`Serde::prepare`], so a schema serde can derive
32/// `<topic>-key` or `<topic>-value`.
33#[derive(Debug, Clone, Copy, PartialEq, Eq)]
34pub enum SerdeRole {
35    Key,
36    Value,
37}
38
39/// Associate a concrete Serde type with its deserialized target type.
40///
41/// This association gives automatic type inference.
42#[diagnostic::on_unimplemented(
43    message = "the serde type `{Self}` is not associated with any target model type",
44    label = "does not implement `SerdeAssociate`",
45    note = "implement `SerdeAssociate` for `{Self}` and define `type Target = ...;` to map this Serde to its model type"
46)]
47pub trait SerdeAssociate {
48    type Target: Send + Sync + 'static;
49}
50
51/// The key and value [`Serde`]s that **read** a topic into the topology.
52///
53/// A source node deserializes incoming bytes with these, and so does
54/// [`TopologyTestDriver::pipe_input`]. One named argument holds both serdes, so
55/// their roles stay visible at the call site.
56/// `Consumed::with(keySerde, valueSerde)` reads the same as Kafka Streams'
57/// `Consumed`.
58///
59/// [`TopologyTestDriver::pipe_input`]: crate::TopologyTestDriver::pipe_input
60#[derive(Debug, Clone, Copy)]
61pub struct Consumed<KS, VS> {
62    pub(crate) key_serde: KS,
63    pub(crate) value_serde: VS,
64}
65
66impl<KS, VS> Consumed<KS, VS> {
67    /// Pair a key serde with a value serde, key first. This matches
68    /// `Consumed.with(keySerde, valueSerde)`.
69    #[must_use]
70    pub fn with(key_serde: KS, value_serde: VS) -> Self {
71        Self {
72            key_serde,
73            value_serde,
74        }
75    }
76}
77
78/// The key and value [`Serde`]s that **write** a topic from the topology.
79///
80/// A sink node serializes outgoing records with these.
81/// [`TopologyTestDriver::read_output`] uses them to deserialize what a sink
82/// wrote. `Produced::with(keySerde, valueSerde)` reads the same as Kafka
83/// Streams' `Produced`.
84///
85/// [`TopologyTestDriver::read_output`]: crate::TopologyTestDriver::read_output
86#[derive(Debug, Clone, Copy)]
87pub struct Produced<KS, VS> {
88    pub(crate) key_serde: KS,
89    pub(crate) value_serde: VS,
90}
91
92impl<KS, VS> Produced<KS, VS> {
93    /// Pair a key serde with a value serde, key first. This matches
94    /// `Produced.with(keySerde, valueSerde)`.
95    #[must_use]
96    pub fn with(key_serde: KS, value_serde: VS) -> Self {
97        Self {
98            key_serde,
99            value_serde,
100        }
101    }
102}
103
104/// A clonable [`Serde<T>`] over a type-erased `Arc<dyn Serde<T>>`.
105///
106/// A serde captured behind an `Arc`, such as a `KTable`'s stored key or value
107/// serde, can go to `add_source` or `add_sink`, which want a `Serde<T> + Clone`
108/// value. It can also go into a processor field as a box. The caller never names
109/// the concrete serde type.
110pub(crate) struct SerdeArc<T>(pub(crate) std::sync::Arc<dyn Serde<T>>);
111
112impl<T> Clone for SerdeArc<T> {
113    fn clone(&self) -> Self {
114        SerdeArc(std::sync::Arc::clone(&self.0))
115    }
116}
117
118impl<T: Send + Sync + 'static> Serde<T> for SerdeArc<T> {
119    fn serialize(&self, topic: &str, value: &T) -> Bytes {
120        self.0.serialize(topic, value)
121    }
122    fn deserialize(&self, topic: &str, bytes: &[u8]) -> Result<T, SerdeError> {
123        self.0.deserialize(topic, bytes)
124    }
125    fn prepare(&self, topic: &str, role: SerdeRole) {
126        self.0.prepare(topic, role);
127    }
128}
129
130impl<T: Send + Sync + 'static> SerdeAssociate for SerdeArc<T> {
131    type Target = T;
132}
133
134/// Identity serde for raw `Bytes`.
135#[derive(Debug, Clone, Copy, Default)]
136pub struct BytesSerde;
137impl Serde<Bytes> for BytesSerde {
138    fn serialize(&self, _topic: &str, value: &Bytes) -> Bytes {
139        value.clone()
140    }
141    fn deserialize(&self, _topic: &str, bytes: &[u8]) -> Result<Bytes, SerdeError> {
142        Ok(Bytes::copy_from_slice(bytes))
143    }
144}
145
146impl SerdeAssociate for BytesSerde {
147    type Target = Bytes;
148}
149
150/// UTF-8 `String` serde.
151#[derive(Debug, Clone, Copy, Default)]
152pub struct StringSerde;
153impl Serde<String> for StringSerde {
154    fn serialize(&self, _topic: &str, value: &String) -> Bytes {
155        Bytes::copy_from_slice(value.as_bytes())
156    }
157    fn deserialize(&self, _topic: &str, bytes: &[u8]) -> Result<String, SerdeError> {
158        String::from_utf8(bytes.to_vec()).map_err(|e| SerdeError(e.to_string()))
159    }
160}
161
162impl SerdeAssociate for StringSerde {
163    type Target = String;
164}
165
166/// Big-endian 8-byte `i64` serde (matches the JVM `Serdes.Long()`).
167#[derive(Debug, Clone, Copy, Default)]
168pub struct I64Serde;
169impl Serde<i64> for I64Serde {
170    fn serialize(&self, _topic: &str, value: &i64) -> Bytes {
171        Bytes::copy_from_slice(&value.to_be_bytes())
172    }
173    fn deserialize(&self, _topic: &str, bytes: &[u8]) -> Result<i64, SerdeError> {
174        let arr: [u8; 8] = bytes
175            .try_into()
176            .map_err(|_| SerdeError(format!("expected 8 bytes, got {}", bytes.len())))?;
177        Ok(i64::from_be_bytes(arr))
178    }
179}
180
181impl SerdeAssociate for I64Serde {
182    type Target = i64;
183}
184
185/// Mapping trait for types that have a default associated [`Serde`].
186///
187/// A type that does not implement this trait cannot use `.stream` or the other
188/// ergonomic methods. The compiler reports that it found no default serde.
189///
190/// # Examples
191///
192/// ```compile_fail
193/// use krabka_client_streams::{StreamsBuilder, KStream};
194///
195/// struct CustomType;
196///
197/// let builder = StreamsBuilder::new();
198/// // Fails because CustomType does not implement DefaultSerde:
199/// let stream: KStream<CustomType, String> = builder.stream(["topic"]);
200/// ```
201#[diagnostic::on_unimplemented(
202    message = "the type `{Self}` does not have a default Serde implementation",
203    label = "no default Serde association found for this type",
204    note = "implement `DefaultSerde` for `{Self}` or specify/override the serde explicitly using `.with_key_serde()` or `.with_value_serde()`"
205)]
206pub trait DefaultSerde: Send + Sync + 'static + Sized {
207    type Serde: SerdeAssociate<Target = Self> + Serde<Self> + Clone + Default + 'static;
208}
209
210impl DefaultSerde for String {
211    type Serde = StringSerde;
212}
213
214impl DefaultSerde for i64 {
215    type Serde = I64Serde;
216}
217
218impl DefaultSerde for bytes::Bytes {
219    type Serde = BytesSerde;
220}
221
222impl<KS, VS> From<(KS, VS)> for Consumed<KS, VS> {
223    fn from((key_serde, value_serde): (KS, VS)) -> Self {
224        Self::with(key_serde, value_serde)
225    }
226}
227
228impl<KS, VS> From<(KS, VS)> for Produced<KS, VS> {
229    fn from((key_serde, value_serde): (KS, VS)) -> Self {
230        Self::with(key_serde, value_serde)
231    }
232}
233
234/// Write a Kafka signed/zigzag varint, the same encoding as the JVM
235/// `ByteUtils.writeVarint`.
236///
237/// This function zigzag-encodes the value with `n << 1 ^ n >> 31` for the 32-bit
238/// equivalent. It then writes a base-128 little-endian varint. A value ≤ 63
239/// encodes to one byte.
240fn write_varint(n: usize, buf: &mut Vec<u8>) {
241    // Zigzag encode: treat as i64 equivalent — all inputs here are non-negative
242    // lengths, so the zigzag is simply n << 1 (sign bit = 0).
243    let mut v = (n as u64) << 1;
244    loop {
245        if v < 0x80 {
246            buf.push(v.to_le_bytes()[0]);
247            break;
248        }
249        buf.push(v.to_le_bytes()[0] | 0x80);
250        v >>= 7;
251    }
252}
253
254/// Read a Kafka signed/zigzag varint.
255///
256/// Returns `(decoded_value, bytes_consumed)`.
257fn read_varint(bytes: &[u8]) -> Result<(usize, usize), SerdeError> {
258    let mut value: u64 = 0;
259    let mut shift = 0u32;
260    let mut consumed = 0;
261    for &b in bytes {
262        consumed += 1;
263        let low7 = u64::from(b & 0x7F);
264        value |= low7 << shift;
265        shift += 7;
266        if b & 0x80 == 0 {
267            // Zigzag decode: (value >>> 1) ^ -(value & 1)
268            let decoded = (value >> 1) ^ ((value & 1).wrapping_neg());
269            let n = usize::try_from(decoded)
270                .map_err(|_| SerdeError(format!("varint value {decoded} out of usize range")))?;
271            return Ok((n, consumed));
272        }
273        if shift >= 64 {
274            return Err(SerdeError("varint overflow".to_string()));
275        }
276    }
277    Err(SerdeError("truncated varint".to_string()))
278}
279
280/// [`Serde`] for `Change<V>`, byte-compatible with the JVM
281/// `ChangedSerializer` and `ChangedDeserializer`. Repartition topics use it when
282/// a `KGroupedTable` re-keys.
283///
284/// # Wire format
285///
286/// | Case | Layout |
287/// |------|--------|
288/// | new only | `new_bytes ++ [0x01]` |
289/// | old only | `old_bytes ++ [0x00]` |
290/// | both     | `varint(new_len) ++ new_bytes ++ old_bytes ++ [0x02]` |
291///
292/// The `varint` is a Kafka signed/zigzag varint (`ByteUtils.writeVarint` / `readVarint`).
293#[derive(Debug, Clone, Copy)]
294pub struct Changed<S> {
295    inner: S,
296}
297
298impl<S> Changed<S> {
299    /// Wrap an inner serde to produce a `Changed<V>` serde.
300    #[must_use]
301    pub fn new(inner: S) -> Self {
302        Self { inner }
303    }
304}
305
306impl<V, S> Serde<crate::dsl::processors::change::Change<V>> for Changed<S>
307where
308    V: Send + Sync + 'static,
309    S: Serde<V>,
310{
311    fn serialize(&self, topic: &str, value: &crate::dsl::processors::change::Change<V>) -> Bytes {
312        match (&value.new, &value.old) {
313            (Some(new), Some(old)) => {
314                let new_bytes = self.inner.serialize(topic, new);
315                let old_bytes = self.inner.serialize(topic, old);
316                let mut buf = Vec::with_capacity(1 + new_bytes.len() + old_bytes.len() + 1);
317                write_varint(new_bytes.len(), &mut buf);
318                buf.extend_from_slice(&new_bytes);
319                buf.extend_from_slice(&old_bytes);
320                buf.push(0x02);
321                Bytes::from(buf)
322            }
323            (Some(new), None) => {
324                let new_bytes = self.inner.serialize(topic, new);
325                let mut buf = Vec::with_capacity(new_bytes.len() + 1);
326                buf.extend_from_slice(&new_bytes);
327                buf.push(0x01);
328                Bytes::from(buf)
329            }
330            (None, Some(old)) => {
331                let old_bytes = self.inner.serialize(topic, old);
332                let mut buf = Vec::with_capacity(old_bytes.len() + 1);
333                buf.extend_from_slice(&old_bytes);
334                buf.push(0x00);
335                Bytes::from(buf)
336            }
337            // The repartition-map only ever forwards combined, subtract-only, or
338            // add-only changes — never a both-`None` change, and the JVM
339            // `ChangedSerializer` likewise rejects an all-null change. Fail loud
340            // rather than silently emit an unsymmetric encoding.
341            (None, None) => unreachable!("Changed serde never carries a both-None change"),
342        }
343    }
344
345    fn deserialize(
346        &self,
347        topic: &str,
348        bytes: &[u8],
349    ) -> Result<crate::dsl::processors::change::Change<V>, SerdeError> {
350        if bytes.is_empty() {
351            return Err(SerdeError("empty bytes for Change".to_string()));
352        }
353        let flag = bytes[bytes.len() - 1];
354        let payload = &bytes[..bytes.len() - 1];
355        match flag {
356            0x00 => {
357                let old = self.inner.deserialize(topic, payload)?;
358                Ok(crate::dsl::processors::change::Change {
359                    old: Some(old),
360                    new: None,
361                })
362            }
363            0x01 => {
364                let new = self.inner.deserialize(topic, payload)?;
365                Ok(crate::dsl::processors::change::Change {
366                    old: None,
367                    new: Some(new),
368                })
369            }
370            0x02 => {
371                let (new_len, consumed) = read_varint(payload)?;
372                let after_varint = &payload[consumed..];
373                if after_varint.len() < new_len {
374                    return Err(SerdeError(format!(
375                        "not enough bytes for new value: need {new_len}, have {}",
376                        after_varint.len()
377                    )));
378                }
379                let new_bytes = &after_varint[..new_len];
380                let old_bytes = &after_varint[new_len..];
381                let new = self.inner.deserialize(topic, new_bytes)?;
382                let old = self.inner.deserialize(topic, old_bytes)?;
383                Ok(crate::dsl::processors::change::Change {
384                    old: Some(old),
385                    new: Some(new),
386                })
387            }
388            other => Err(SerdeError(format!("unknown Change flag: 0x{other:02x}"))),
389        }
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use assert2::check;
396
397    use super::*;
398
399    #[test]
400    fn string_serde_round_trips() {
401        let s = StringSerde;
402        let b = s.serialize("t", &"héllo".to_string());
403        check!(s.deserialize("t", &b).unwrap() == "héllo");
404    }
405
406    #[test]
407    fn i64_serde_is_big_endian_8_bytes() {
408        let s = I64Serde;
409        let b = s.serialize("t", &1i64);
410        check!(b.as_ref() == [0, 0, 0, 0, 0, 0, 0, 1]);
411        check!(s.deserialize("t", &b).unwrap() == 1);
412        check!(s.deserialize("t", &[0, 1]).is_err());
413    }
414
415    #[test]
416    fn bytes_serde_is_identity() {
417        let s = BytesSerde;
418        let b = s.serialize("t", &bytes::Bytes::from_static(b"xy"));
419        check!(s.deserialize("t", &b).unwrap() == bytes::Bytes::from_static(b"xy"));
420    }
421
422    fn hex(b: &[u8]) -> String {
423        b.iter().fold(String::new(), |mut s, x| {
424            use std::fmt::Write as _;
425            write!(s, "{x:02x}").unwrap();
426            s
427        })
428    }
429
430    fn golden() -> serde_json::Value {
431        // Resolved while the test runs, not while it compiles. `env!` bakes the
432        // absolute build directory into the binary, so the same sources would
433        // produce different bytes on different machines and a sandboxed build
434        // refuses them. Cargo sets the variable for test binaries; a build that
435        // does not runs from a root where the crate path locates the file.
436        let base = std::env::var("CARGO_MANIFEST_DIR")
437            .unwrap_or_else(|_| "crates/client-streams".to_owned());
438        let path =
439            std::path::Path::new(&base).join("tests/testdata/kgrouped_table/changed_bytes.json");
440        let raw = std::fs::read_to_string(&path).unwrap_or_else(|error| {
441            panic!("read changed_bytes golden {}: {error}", path.display())
442        });
443        serde_json::from_str(&raw).unwrap()
444    }
445
446    #[test]
447    fn changed_long_matches_jvm_bytes() {
448        let g = golden();
449        let s = Changed::new(I64Serde);
450        let both = crate::dsl::processors::change::Change {
451            old: Some(2i64),
452            new: Some(6i64),
453        };
454        let new_only = crate::dsl::processors::change::Change {
455            old: None,
456            new: Some(5i64),
457        };
458        let old_only = crate::dsl::processors::change::Change {
459            old: Some(4i64),
460            new: None,
461        };
462        check!(hex(&s.serialize("topic", &both)) == g["both"].as_str().unwrap());
463        check!(hex(&s.serialize("topic", &new_only)) == g["new_only"].as_str().unwrap());
464        check!(hex(&s.serialize("topic", &old_only)) == g["old_only"].as_str().unwrap());
465    }
466
467    #[test]
468    fn changed_round_trips() {
469        let s = Changed::new(I64Serde);
470        for c in [
471            crate::dsl::processors::change::Change {
472                old: Some(2i64),
473                new: Some(6i64),
474            },
475            crate::dsl::processors::change::Change {
476                old: None,
477                new: Some(5i64),
478            },
479            crate::dsl::processors::change::Change {
480                old: Some(4i64),
481                new: None,
482            },
483        ] {
484            let bytes = s.serialize("topic", &c);
485            let back: crate::dsl::processors::change::Change<i64> =
486                s.deserialize("topic", &bytes).unwrap();
487            check!(back == c);
488        }
489    }
490}
krabka_client_streams/processor/serde.rs

krabka_client_streams/processor/
serde.rs
