crabka_connect/
source.rs

1//! The [`Source`] SPI: pull records out of an external system, with a
2//! checkpointable read position.
3
4use async_trait::async_trait;
5
6use crate::{
7    error::ConnectError,
8    record::{ConnectRecord, SourceOffset},
9};
10
11/// A connector that pulls records out of an external system (a database change
12/// stream, a file tail, a queue) for production into Kafka.
13///
14/// This is the read side of the connector SPI and the template every CDC source
15/// builds on. It mirrors the streams runtime's `RecordFetcher`, but pulls one
16/// record at a time and owns its own read position rather than being told an
17/// offset on every call.
18///
19/// ## Polling
20///
21/// [`poll`](Source::poll) returns the next record, or `None` when the source is
22/// momentarily caught up (the runtime should back off and poll again). Each
23/// successful `poll` advances the source's internal position; the runtime never
24/// passes an offset in — it reads the position back with
25/// [`checkpoint`](Source::checkpoint).
26///
27/// ## Offset state
28///
29/// The runtime persists the [`SourceOffset`] returned by `checkpoint` after the
30/// records it covers have been durably produced, and restores it with
31/// [`seek`](Source::seek) before the first `poll` on restart. The offset is
32/// opaque to the runtime — only the source interprets it — so a source is free
33/// to encode a log sequence number, a byte offset, a GTID set, or whatever
34/// resume token its backend uses.
35///
36/// After the sink commit is durable and checkpoint persistence succeeds, the
37/// runtime calls [`acknowledge`](Source::acknowledge) with the persisted offset.
38/// Sources that hold backend resources such as non-advancing cursors or logical
39/// replication slots can use this hook to release data that is now safe to
40/// discard. The default implementation is a no-op for sources that need no
41/// explicit acknowledgement.
42#[async_trait]
43pub trait Source<K, V>: Send + Sync + 'static {
44    /// Pull the next record, advancing the source's read position.
45    ///
46    /// Returns `Ok(None)` when nothing is available yet — a non-fatal "caught
47    /// up" signal, not end-of-stream. Returns `Err` only on a real failure
48    /// (lost connection, malformed upstream data the source cannot skip).
49    ///
50    /// # Errors
51    ///
52    /// Returns [`ConnectError`] if reading from the external system fails.
53    async fn poll(&mut self) -> Result<Option<ConnectRecord<K, V>>, ConnectError>;
54
55    /// Snapshot the current read position so the runtime can persist it.
56    ///
57    /// Returns `None` before the source has a position to commit (e.g. nothing
58    /// has been polled and no prior offset was restored). The runtime commits a
59    /// checkpoint only after the records preceding it are durable, so on restart
60    /// [`seek`](Source::seek) resumes from the last fully-produced record.
61    fn checkpoint(&self) -> Option<SourceOffset>;
62
63    /// Restore the read position to a previously [`checkpoint`](Source::checkpoint)ed
64    /// offset. Called once, before the first [`poll`](Source::poll), on startup
65    /// and after a rebalance. A source with no stored offset is not sought and
66    /// starts from its configured default position.
67    ///
68    /// # Errors
69    ///
70    /// Returns [`ConnectError::Offset`] if `offset` does not name a position
71    /// this source can resume from (e.g. the upstream log has since been
72    /// truncated past it).
73    async fn seek(&mut self, offset: SourceOffset) -> Result<(), ConnectError>;
74
75    /// Acknowledge that `offset` is durable end-to-end.
76    ///
77    /// The runtime calls this only after a non-empty batch has been committed
78    /// to the sink and the same offset has been saved in the checkpoint store.
79    /// It is never called before checkpoint persistence, so a failing
80    /// checkpoint save prevents upstream acknowledgement. Sources may use this
81    /// to advance external cursors or release retained log segments.
82    ///
83    /// The default is a no-op, preserving existing source implementations.
84    ///
85    /// # Errors
86    ///
87    /// Returns [`ConnectError`] if the upstream acknowledgement fails.
88    async fn acknowledge(&mut self, _offset: &SourceOffset) -> Result<(), ConnectError> {
89        Ok(())
90    }
91
92    /// Release any resources held by the source (connections, file handles).
93    /// The default is a no-op. After `close`, the source is not polled again.
94    ///
95    /// # Errors
96    ///
97    /// Returns [`ConnectError`] if cleanup fails.
98    async fn close(&mut self) -> Result<(), ConnectError> {
99        Ok(())
100    }
101}
crabka_connect/source.rs

crabka_connect/
source.rs
