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

crabka_connect/
source.rs
