krabka_client_streams/runtime/io.rs
1//! The dependency-injected I/O that the runtime needs: it fetches source
2//! records, produces sink records, and commits and fetches offsets.
3//!
4//! The real broker implementations live in `io_broker.rs`. Fakes in the tests
5//! make `StreamTask` and `StreamThread` testable and deterministic without a
6//! broker.
7
8use bytes::Bytes;
9
10use crate::error::StreamsClientError;
11
12/// A fetched source record. The timestamp is `-1` when the fetcher cannot supply
13/// it.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct FetchedRec {
16 pub offset: i64,
17 pub key: Option<Bytes>,
18 pub value: Option<Bytes>,
19 pub timestamp: i64,
20}
21
22/// A batch of consecutive records from one partition.
23#[derive(Debug, Clone, Default)]
24pub struct FetchBatch {
25 pub records: Vec<FetchedRec>,
26}
27
28impl FetchBatch {
29 /// The offset to fetch next. It is one past the last record, or `current`
30 /// when the batch is empty.
31 #[must_use]
32 pub fn next_offset(&self, current: i64) -> i64 {
33 self.records.last().map_or(current, |r| r.offset + 1)
34 }
35}
36
37/// Fetch isolation level, the Kafka `Fetch.isolation_level`.
38///
39/// Under EOS-v2 the changelog restore must read `ReadCommitted`, so that it
40/// excludes aborted writes. An aborted write is a record produced inside a
41/// transaction that later aborted. The restored store then holds only committed
42/// state. Normal source processing and the global-store bootstrap use
43/// `ReadUncommitted`.
44#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
45pub enum IsolationLevel {
46 #[default]
47 ReadUncommitted,
48 ReadCommitted,
49}
50
51#[async_trait::async_trait]
52pub trait RecordFetcher: Send + Sync + 'static {
53 /// Fetch the records for `(topic, partition)` from `offset`, at the given
54 /// `isolation` level. An empty batch means that nothing new is available.
55 async fn fetch(
56 &self,
57 topic: &str,
58 partition: i32,
59 offset: i64,
60 isolation: IsolationLevel,
61 ) -> Result<FetchBatch, StreamsClientError>;
62
63 /// The partition indices of `topic`. The global consumer reads all of them
64 /// to materialize a fully-replicated global store. Default: the single
65 /// partition 0. The broker fetcher overrides the default from the
66 /// metadata.
67 async fn partitions(&self, _topic: &str) -> Result<Vec<i32>, StreamsClientError> {
68 Ok(vec![0])
69 }
70}
71
72#[async_trait::async_trait]
73pub trait RecordProducer: Send + Sync + 'static {
74 /// Enqueue a record to `topic`.
75 ///
76 /// `partition`:
77 /// - `None` uses the producer's key-hash partitioner. This is correct for a
78 /// sink topic or a repartition topic.
79 /// - `Some(p)` pins the record to partition `p`. A changelog topic needs
80 /// this, so that the task restore can read the record back from the task
81 /// partition.
82 async fn send(
83 &self,
84 topic: &str,
85 partition: Option<i32>,
86 key: Option<Bytes>,
87 value: Option<Bytes>,
88 ) -> Result<(), StreamsClientError>;
89 /// Like `send`, but it sets the record's timestamp in epoch milliseconds
90 /// when `timestamp_ms` is `Some`.
91 ///
92 /// The default delegates to `send`, and the producer then fills in the
93 /// wall-clock time. The broker producer overrides this method, so that a
94 /// versioned-store changelog record carries the version timestamp
95 /// (KIP-889).
96 async fn send_with_timestamp(
97 &self,
98 topic: &str,
99 partition: Option<i32>,
100 key: Option<Bytes>,
101 value: Option<Bytes>,
102 _timestamp_ms: Option<i64>,
103 ) -> Result<(), StreamsClientError> {
104 self.send(topic, partition, key, value).await
105 }
106 /// Block until the broker acknowledges every enqueued record. This is the
107 /// durability barrier.
108 async fn flush(&self) -> Result<(), StreamsClientError>;
109}
110
111/// The lazy begin-the-transaction gate passed to the stream task's
112/// `process_once` operation under EOS-v2.
113///
114/// The task calls [`BeginTxnGate::ensure_begun`] exactly before its first
115/// produced record in a commit interval. An interval that fetches no records
116/// therefore opens no transaction, which avoids empty-txn churn on an idle app.
117/// Under at-least-once the caller passes no gate.
118#[async_trait::async_trait]
119pub trait BeginTxnGate: Send {
120 /// Make sure a transaction is open. The first call within the interval
121 /// begins one, and every later call does nothing. The task calls this method
122 /// right before the first sink or changelog `send`.
123 async fn ensure_begun(&mut self) -> Result<(), StreamsClientError>;
124}
125
126#[async_trait::async_trait]
127pub trait OffsetStore: Send + Sync + 'static {
128 /// The committed offset for `(topic, partition)`, or `None` when no commit
129 /// happened.
130 async fn committed(
131 &self,
132 topic: &str,
133 partition: i32,
134 ) -> Result<Option<i64>, StreamsClientError>;
135 /// The earliest available offset, which matches
136 /// `auto.offset.reset = earliest`.
137 async fn earliest(&self, topic: &str, partition: i32) -> Result<i64, StreamsClientError>;
138 /// The latest available offset, that is the log-end offset.
139 async fn latest(&self, topic: &str, partition: i32) -> Result<i64, StreamsClientError>;
140 /// Commit `(topic, partition, offset)` triples for the streams group.
141 async fn commit(&self, offsets: &[(String, i32, i64)]) -> Result<(), StreamsClientError>;
142}
143
144#[cfg(test)]
145mod tests {
146 use assert2::check;
147
148 use super::*;
149
150 #[test]
151 fn fetch_batch_next_offset_advances_past_last() {
152 let b = FetchBatch {
153 records: vec![
154 FetchedRec {
155 offset: 5,
156 key: None,
157 value: Some(bytes::Bytes::from_static(b"a")),
158 timestamp: -1,
159 },
160 FetchedRec {
161 offset: 6,
162 key: None,
163 value: Some(bytes::Bytes::from_static(b"b")),
164 timestamp: -1,
165 },
166 ],
167 };
168 check!(b.next_offset(0) == 7);
169 let empty = FetchBatch { records: vec![] };
170 check!(empty.next_offset(9) == 9);
171 }
172}