crabka_connect/
lib.rs

1//! Connector-framework SPI for Crabka.
2//!
3//! This crate defines the keystone traits that every connector — every CDC
4//! source, every telemetry sink — builds on, plus the converter layer that
5//! bridges typed payloads to the Kafka wire.
6//!
7//! - [`Source`] pulls records out of an external system one at a time
8//!   ([`poll`](Source::poll)), snapshots its read position
9//!   ([`checkpoint`](Source::checkpoint)), and restores it on restart
10//!   ([`seek`](Source::seek)). Sources can optionally acknowledge a persisted
11//!   checkpoint after sink commit with [`acknowledge`](Source::acknowledge).
12//! - [`Sink`] pushes records into an external system in batches
13//!   ([`put`](Sink::put)), makes them durable ([`flush`](Sink::flush)), and
14//!   optionally gates writes behind a transaction for exactly-once delivery
15//!   ([`begin`](Sink::begin) / [`commit`](Sink::commit) / [`abort`](Sink::abort)).
16//! - [`Converter`] bridges a connector's typed payload `T` to and from wire
17//!   [`bytes::Bytes`]. [`ByteIdentity`] is a byte-for-byte passthrough;
18//!   [`SchemaConverter`] wraps the Confluent schema-registry serdes from
19//!   `crabka-schema-serde`.
20//!
21//! - [`ConnectorRuntime`] is the embeddable, single-process driver that owns a
22//!   `Source` + `Sink` pair and pipes records between them — polling, applying
23//!   bounded backpressure, coordinating source checkpoints with sink commits
24//!   through the transactional gate, and exposing pause/resume + graceful-drain
25//!   lifecycle hooks. No Connect worker protocol, no REST.
26//!
27//! The transport traits mirror the streams runtime I/O traits
28//! (`RecordFetcher` / `RecordProducer`) and the remote-storage
29//! `RemoteStorageManager` SPI; the converter layer mirrors Kafka Connect's
30//! `Converter`. All records carry an optional key, optional value (a `None`
31//! value is a tombstone), an optional timestamp, and ordered headers — see
32//! [`ConnectRecord`].
33
34pub mod config;
35pub mod convert;
36pub mod error;
37pub mod ids;
38pub mod record;
39pub mod runtime;
40pub mod sink;
41pub mod source;
42
43pub use config::{
44    ConfigDef, ConfigError, ConfigKey, ConfigKind, ConfigResult, ConnectorConfig,
45    EnvSecretResolver, FromResolvedValue, RawConfig, ResolveOptions, ResolvedConfig, SecretRef,
46    SecretResolutionError, SecretResolver, SecretString,
47};
48pub use convert::{ByteIdentity, Converter, SchemaConverter};
49#[cfg(feature = "derive")]
50pub use crabka_connect_derive::ConnectorConfig;
51pub use error::ConnectError;
52pub use ids::{PartitionMap, PositionMap};
53pub use record::{ConnectRecord, Header, OffsetMap, OffsetValue, SourceOffset};
54pub use runtime::{
55    CheckpointStore, ConnectorHandle, ConnectorRuntime, HasSink, HasSource,
56    InMemoryCheckpointStore, NoSink, NoSource, RuntimeState,
57};
58#[doc(hidden)]
59pub use serde_json as __serde_json;
60pub use sink::Sink;
61pub use source::Source;
62
63#[cfg(test)]
64mod tests {
65    //! An end-to-end exercise of the SPI with in-memory fakes, proving the
66    //! traits are object-safe and compose as the runtime would drive them.
67
68    use async_trait::async_trait;
69    use bytes::Bytes;
70
71    use super::*;
72
73    /// A finite in-memory source over `(key, value)` pairs that tracks its read
74    /// position as the index into the backing vector.
75    struct VecSource {
76        records: Vec<(Bytes, Bytes)>,
77        pos: usize,
78    }
79
80    #[async_trait]
81    impl Source<Bytes, Bytes> for VecSource {
82        async fn poll(&mut self) -> Result<Option<ConnectRecord<Bytes, Bytes>>, ConnectError> {
83            let Some((k, v)) = self.records.get(self.pos).cloned() else {
84                return Ok(None);
85            };
86            self.pos += 1;
87            Ok(Some(ConnectRecord::new(Some(k), Some(v))))
88        }
89
90        fn checkpoint(&self) -> Option<SourceOffset> {
91            if self.pos == 0 {
92                return None;
93            }
94            let mut position = OffsetMap::new();
95            #[allow(clippy::cast_possible_wrap)]
96            position.insert("index".into(), OffsetValue::Long(self.pos as i64));
97            Some(SourceOffset::new(OffsetMap::new().into(), position.into()))
98        }
99
100        async fn seek(&mut self, offset: SourceOffset) -> Result<(), ConnectError> {
101            match offset.position.get("index") {
102                Some(OffsetValue::Long(i)) => {
103                    self.pos = usize::try_from(*i)
104                        .map_err(|_| ConnectError::Offset(format!("negative index {i}")))?;
105                    Ok(())
106                }
107                _ => Err(ConnectError::Offset("missing `index` component".into())),
108            }
109        }
110    }
111
112    /// A transactional in-memory sink: `put` stages records, `commit` promotes
113    /// the staged batch into the committed log, `abort` drops it.
114    #[derive(Default)]
115    struct VecSink {
116        staged: Vec<ConnectRecord<Bytes, Bytes>>,
117        committed: Vec<ConnectRecord<Bytes, Bytes>>,
118    }
119
120    #[async_trait]
121    impl Sink<Bytes, Bytes> for VecSink {
122        async fn put(
123            &mut self,
124            records: Vec<ConnectRecord<Bytes, Bytes>>,
125        ) -> Result<(), ConnectError> {
126            self.staged.extend(records);
127            Ok(())
128        }
129
130        async fn flush(&mut self) -> Result<(), ConnectError> {
131            self.committed.append(&mut self.staged);
132            Ok(())
133        }
134
135        fn supports_transactions(&self) -> bool {
136            true
137        }
138
139        async fn abort(&mut self) -> Result<(), ConnectError> {
140            self.staged.clear();
141            Ok(())
142        }
143    }
144
145    #[tokio::test]
146    async fn source_drains_then_checkpoint_and_seek_resume() {
147        use assert2::check;
148
149        let mut src = VecSource {
150            records: vec![
151                (Bytes::from_static(b"k0"), Bytes::from_static(b"v0")),
152                (Bytes::from_static(b"k1"), Bytes::from_static(b"v1")),
153            ],
154            pos: 0,
155        };
156        check!(src.checkpoint().is_none());
157
158        let first = src.poll().await.unwrap().unwrap();
159        check!(first.key == Some(Bytes::from_static(b"k0")));
160        let cp = src.checkpoint().expect("position after one poll");
161
162        // A fresh source seeks to the checkpoint and resumes at the next record.
163        let mut resumed = VecSource {
164            records: src.records.clone(),
165            pos: 0,
166        };
167        resumed.seek(cp).await.unwrap();
168        let next = resumed.poll().await.unwrap().unwrap();
169        check!(next.key == Some(Bytes::from_static(b"k1")));
170        // Drained: further polls report caught-up, not an error.
171        check!(resumed.poll().await.unwrap().is_none());
172    }
173
174    #[tokio::test]
175    async fn sink_commit_promotes_and_abort_discards() {
176        use assert2::check;
177
178        let mut sink = VecSink::default();
179        check!(sink.supports_transactions());
180
181        // A committed interval lands in the log.
182        sink.begin().await.unwrap();
183        sink.put(vec![ConnectRecord::new(
184            None,
185            Some(Bytes::from_static(b"a")),
186        )])
187        .await
188        .unwrap();
189        sink.commit().await.unwrap();
190        check!(sink.committed.len() == 1);
191
192        // An aborted interval is discarded.
193        sink.begin().await.unwrap();
194        sink.put(vec![ConnectRecord::new(
195            None,
196            Some(Bytes::from_static(b"b")),
197        )])
198        .await
199        .unwrap();
200        sink.abort().await.unwrap();
201        check!(sink.committed.len() == 1);
202        check!(sink.staged.is_empty());
203    }
204
205    #[tokio::test]
206    async fn spi_is_object_safe() {
207        use assert2::check;
208
209        // The runtime holds connectors behind trait objects; confirm both
210        // traits are dyn-compatible.
211        let mut src: Box<dyn Source<Bytes, Bytes>> = Box::new(VecSource {
212            records: vec![(Bytes::from_static(b"k"), Bytes::from_static(b"v"))],
213            pos: 0,
214        });
215        let mut sink: Box<dyn Sink<Bytes, Bytes>> = Box::new(VecSink::default());
216
217        while let Some(rec) = src.poll().await.unwrap() {
218            sink.put(vec![rec]).await.unwrap();
219        }
220        sink.flush().await.unwrap();
221
222        let conv: &dyn Converter<Bytes> = &ByteIdentity;
223        let wire = conv.serialize("t", &Bytes::from_static(b"v")).unwrap();
224        check!(wire == Bytes::from_static(b"v"));
225    }
226
227    #[tokio::test]
228    async fn source_seek_rejects_unresumable_offset() {
229        use assert2::check;
230
231        let mut src = VecSource {
232            records: vec![],
233            pos: 0,
234        };
235        // An offset with no `index` component names no position to resume from.
236        check!(src.seek(SourceOffset::default()).await.is_err());
237        // A negative index cannot map to a vector position.
238        let mut position = OffsetMap::new();
239        position.insert("index".into(), OffsetValue::Long(-1));
240        check!(
241            src.seek(SourceOffset::new(OffsetMap::new().into(), position.into()))
242                .await
243                .is_err()
244        );
245    }
246
247    #[tokio::test]
248    async fn source_and_sink_close_are_noops() {
249        use assert2::check;
250
251        let mut src = VecSource {
252            records: vec![],
253            pos: 0,
254        };
255        check!(src.close().await.is_ok());
256        let mut sink = VecSink::default();
257        check!(sink.close().await.is_ok());
258    }
259}
crabka_connect/lib.rs

crabka_connect/
lib.rs
