crabka_connect/record.rs
1//! The record and offset types that flow across the connector SPI.
2
3use std::collections::BTreeMap;
4
5use bytes::Bytes;
6use serde::{Deserialize, Serialize};
7
8use crate::ids::{PartitionMap, PositionMap};
9
10/// A single record crossing the connector boundary, parameterised over its key
11/// and value payload types.
12///
13/// A [`Source`](crate::Source) yields these from `poll`; a [`Sink`](crate::Sink)
14/// consumes them in `put`. For a byte-identity connector `K` and `V` are
15/// [`Bytes`]; for a schema-aware connector they are the connector's domain
16/// types, bridged to the wire by a [`Converter`](crate::Converter).
17///
18/// Both `key` and `value` are optional: a `None` key matches Kafka's null key,
19/// and a `None` value is a tombstone (KIP-87 / compacted-topic delete).
20#[derive(Debug, Clone, PartialEq, Eq, Default)]
21pub struct ConnectRecord<K, V> {
22 /// The record key, or `None` for a null key.
23 pub key: Option<K>,
24 /// The record value, or `None` for a tombstone.
25 pub value: Option<V>,
26 /// The record timestamp in epoch milliseconds, or `None` when the source
27 /// cannot surface one (the producer fills wall-clock time, matching
28 /// `RecordProducer::send` in the streams runtime).
29 pub timestamp: Option<i64>,
30 /// Per-record headers, in declaration order. Header keys may repeat, so
31 /// this is a `Vec`, not a map (matching Kafka's `RecordHeaders`).
32 pub headers: Vec<Header>,
33}
34
35impl<K, V> ConnectRecord<K, V> {
36 /// A record carrying just a key and value, with no timestamp or headers.
37 pub fn new(key: Option<K>, value: Option<V>) -> Self {
38 Self {
39 key,
40 value,
41 timestamp: None,
42 headers: Vec::new(),
43 }
44 }
45
46 /// Set the record timestamp (epoch millis).
47 #[must_use]
48 pub fn with_timestamp(mut self, timestamp_ms: i64) -> Self {
49 self.timestamp = Some(timestamp_ms);
50 self
51 }
52
53 /// Append a header.
54 #[must_use]
55 pub fn with_header(mut self, key: impl Into<String>, value: Option<Bytes>) -> Self {
56 self.headers.push(Header {
57 key: key.into(),
58 value,
59 });
60 self
61 }
62}
63
64/// A single Kafka record header: a string key and an opaque optional value.
65///
66/// Mirrors `crabka_protocol::records::RecordHeader` and the producer's `Header`;
67/// kept here so the connector SPI does not pull in the wire-record crates.
68#[derive(Debug, Clone, PartialEq, Eq, Default)]
69pub struct Header {
70 /// The header key. Not required to be unique within a record.
71 pub key: String,
72 /// The header value, or `None` for a null-valued header.
73 pub value: Option<Bytes>,
74}
75
76/// A connector-defined source position: *which* stream, and *where* in it.
77///
78/// The framework treats both maps as opaque — it persists a checkpoint verbatim
79/// and hands it back to [`Source::seek`](crate::Source::seek) on restart; only
80/// the connector interprets the contents. Mirrors Kafka Connect's split of a
81/// `SourceRecord` into a `sourcePartition` and a `sourceOffset`.
82#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
83pub struct SourceOffset {
84 /// Identifies the stream this position belongs to — e.g. a database table,
85 /// a file path, a log shard. Mirrors Kafka Connect's `sourcePartition`.
86 pub partition: PartitionMap,
87 /// The position within that stream — e.g. a log sequence number, a byte
88 /// offset, a row id. Mirrors Kafka Connect's `sourceOffset`.
89 pub position: PositionMap,
90}
91
92impl SourceOffset {
93 /// A source offset from a stream-identifying `partition` and a `position`
94 /// within it.
95 ///
96 /// The two halves are distinct types ([`PartitionMap`] / [`PositionMap`])
97 /// so the compiler rejects a transposed call — both are the same underlying
98 /// [`OffsetMap`], and a swap here would silently corrupt every checkpoint.
99 #[must_use]
100 pub fn new(partition: PartitionMap, position: PositionMap) -> Self {
101 Self {
102 partition,
103 position,
104 }
105 }
106}
107
108/// A connector offset map: structured, JSON-serialisable key/value pairs.
109/// Ordered (`BTreeMap`) so a serialised checkpoint is byte-stable.
110pub type OffsetMap = BTreeMap<String, OffsetValue>;
111
112/// A scalar value in an [`OffsetMap`]. The variants cover the JSON-compatible
113/// primitives Kafka Connect offsets are built from, so a checkpoint round-trips
114/// through any JSON-based offset store.
115#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
116#[serde(untagged)]
117pub enum OffsetValue {
118 /// A missing / explicitly-null position component.
119 Null,
120 /// A boolean flag.
121 Bool(bool),
122 /// A 64-bit signed integer (LSN, byte offset, row id, …).
123 Long(i64),
124 /// A 64-bit float.
125 Double(f64),
126 /// A UTF-8 string (GTID, snapshot name, opaque cursor token, …).
127 String(String),
128}
129
130impl From<i64> for OffsetValue {
131 fn from(v: i64) -> Self {
132 OffsetValue::Long(v)
133 }
134}
135
136impl From<bool> for OffsetValue {
137 fn from(v: bool) -> Self {
138 OffsetValue::Bool(v)
139 }
140}
141
142impl From<f64> for OffsetValue {
143 fn from(v: f64) -> Self {
144 OffsetValue::Double(v)
145 }
146}
147
148impl From<String> for OffsetValue {
149 fn from(v: String) -> Self {
150 OffsetValue::String(v)
151 }
152}
153
154impl From<&str> for OffsetValue {
155 fn from(v: &str) -> Self {
156 OffsetValue::String(v.to_owned())
157 }
158}
159
160#[cfg(test)]
161mod tests {
162 use assert2::check;
163
164 use super::*;
165
166 #[test]
167 fn builder_sets_timestamp_and_headers() {
168 let r = ConnectRecord::new(
169 Some(Bytes::from_static(b"k")),
170 Some(Bytes::from_static(b"v")),
171 )
172 .with_timestamp(42)
173 .with_header("h", Some(Bytes::from_static(b"hv")));
174 check!(r.timestamp == Some(42));
175 check!(r.headers.len() == 1);
176 check!(r.headers[0].key == "h");
177 check!(r.headers[0].value == Some(Bytes::from_static(b"hv")));
178 }
179
180 #[test]
181 fn tombstone_value_is_none() {
182 let r: ConnectRecord<Bytes, Bytes> =
183 ConnectRecord::new(Some(Bytes::from_static(b"k")), None);
184 check!(r.value.is_none());
185 }
186
187 #[test]
188 fn source_offset_round_trips_through_json() {
189 let mut partition = OffsetMap::new();
190 partition.insert("table".into(), "public.orders".into());
191 let mut position = OffsetMap::new();
192 position.insert("lsn".into(), 12_345_i64.into());
193 position.insert("snapshot".into(), OffsetValue::Bool(false));
194 let offset = SourceOffset::new(partition.into(), position.into());
195
196 let json = serde_json::to_string(&offset).expect("serialize");
197 // The `PartitionMap`/`PositionMap` newtypes are `#[serde(transparent)]`,
198 // so each half encodes as the bare map — no wrapping object. A persisted
199 // checkpoint's byte shape is exactly the two `sourcePartition`/
200 // `sourceOffset` maps Kafka Connect writes.
201 check!(
202 json == r#"{"partition":{"table":"public.orders"},"position":{"lsn":12345,"snapshot":false}}"#
203 );
204 let back: SourceOffset = serde_json::from_str(&json).expect("deserialize");
205 check!(back == offset);
206 }
207
208 #[test]
209 fn offset_value_from_conversions() {
210 check!(OffsetValue::from(7_i64) == OffsetValue::Long(7));
211 check!(OffsetValue::from(true) == OffsetValue::Bool(true));
212 check!(OffsetValue::from(1.5_f64) == OffsetValue::Double(1.5));
213 check!(OffsetValue::from("owned".to_owned()) == OffsetValue::String("owned".into()));
214 check!(OffsetValue::from("borrowed") == OffsetValue::String("borrowed".into()));
215 }
216
217 #[test]
218 fn offset_map_is_byte_stable_across_insertion_order() {
219 // BTreeMap ordering makes a serialized checkpoint deterministic
220 // regardless of the order the connector recorded components.
221 let mut a = OffsetMap::new();
222 a.insert("b".into(), 2_i64.into());
223 a.insert("a".into(), 1_i64.into());
224 let mut b = OffsetMap::new();
225 b.insert("a".into(), 1_i64.into());
226 b.insert("b".into(), 2_i64.into());
227 check!(serde_json::to_string(&a).unwrap() == serde_json::to_string(&b).unwrap());
228 }
229}