krabka_client_streams/barrier/
align.rs

1//! Barrier configuration, the barrier a runtime reached, and the callback.
2
3use std::sync::Arc;
4
5use crate::{
6    barrier::cut::BarrierCut, runtime::iqv2::request::Position, store::snapshot::SnapshotStore,
7};
8
9/// The barrier a stream thread reached: the cut, the tasks it snapshotted, and
10/// the offsets it committed.
11///
12/// Every committed offset is a marker offset of the cut, so the committed
13/// position of each aligned partition **is** the cut.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct Barrier {
16    /// The cut the thread aligned on.
17    pub cut: BarrierCut,
18    /// The tasks whose stores the thread snapshotted, as `<subtopology>-<partition>`
19    /// and in ascending order.
20    pub tasks: Vec<String>,
21    /// The offsets the thread committed, as topic to partition to offset.
22    pub offsets: Position,
23}
24
25/// Receives a barrier after a stream thread commits at a cut.
26///
27/// The thread calls the listener once per epoch. The snapshot of every task is
28/// already in the snapshot store, and the committed position of every aligned
29/// partition is the cut. The call runs on the thread that drives the runtime, so
30/// keep the work short.
31///
32/// Any `Fn(&Barrier)` that is `Send + Sync + 'static` is a listener.
33///
34/// # Examples
35///
36/// ```
37/// use std::sync::Arc;
38///
39/// use krabka_client_streams::barrier::{Barrier, BarrierListener};
40///
41/// let listener: Arc<dyn BarrierListener> =
42///     Arc::new(|barrier: &Barrier| println!("aligned on epoch {}", barrier.cut.epoch));
43/// ```
44pub trait BarrierListener: Send + Sync + 'static {
45    /// Reports that every assigned partition reached the cut.
46    fn on_barrier(&self, barrier: &Barrier);
47}
48
49impl<F> BarrierListener for F
50where
51    F: Fn(&Barrier) + Send + Sync + 'static,
52{
53    fn on_barrier(&self, barrier: &Barrier) {
54        self(barrier);
55    }
56}
57
58/// Tells a stream thread which barrier group to align on.
59///
60/// Pass one to [`StreamsApp`](crate::StreamsApp) or to
61/// [`KafkaStreams`](crate::KafkaStreams). The thread then reads the group's cuts
62/// from `__barrier_state`, holds back every record at or above each partition's
63/// marker offset, and fires the barrier when every assigned partition reaches
64/// the cut. At that point it snapshots each task under the cut's epoch, commits
65/// the cut offsets, and calls the listener.
66///
67/// # Examples
68///
69/// ```
70/// use std::sync::Arc;
71///
72/// use krabka_client_streams::{
73///     barrier::{Barrier, BarrierAlignment},
74///     store::snapshot::FileSnapshotStore,
75/// };
76///
77/// let snapshots = Arc::new(FileSnapshotStore::new("/var/lib/app/state"));
78/// let alignment = BarrierAlignment::on("transactions", snapshots).with_listener(Arc::new(
79///     |barrier: &Barrier| {
80///         println!("epoch {}", barrier.cut.epoch);
81///     },
82/// ));
83/// ```
84#[derive(Clone)]
85pub struct BarrierAlignment {
86    group: String,
87    snapshots: Arc<dyn SnapshotStore>,
88    listener: Option<Arc<dyn BarrierListener>>,
89}
90
91impl BarrierAlignment {
92    /// Aligns on the cuts of `group` and keeps each cut's snapshot in
93    /// `snapshots`.
94    #[must_use]
95    pub fn on(group: impl Into<String>, snapshots: Arc<dyn SnapshotStore>) -> Self {
96        Self {
97            group: group.into(),
98            snapshots,
99            listener: None,
100        }
101    }
102
103    /// Adds the callback the thread calls after each barrier commit.
104    #[must_use]
105    pub fn with_listener(mut self, listener: Arc<dyn BarrierListener>) -> Self {
106        self.listener = Some(listener);
107        self
108    }
109
110    /// The barrier group this alignment reads.
111    #[must_use]
112    pub fn group(&self) -> &str {
113        &self.group
114    }
115
116    /// The store that keeps each cut's snapshot.
117    #[must_use]
118    pub fn snapshots(&self) -> &Arc<dyn SnapshotStore> {
119        &self.snapshots
120    }
121
122    /// The callback the thread calls after each barrier commit, when one is
123    /// set.
124    #[must_use]
125    pub fn listener(&self) -> Option<&Arc<dyn BarrierListener>> {
126        self.listener.as_ref()
127    }
128}
129
130impl std::fmt::Debug for BarrierAlignment {
131    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
132        // The snapshot store is a trait object with no `Debug` bound.
133        f.debug_struct("BarrierAlignment")
134            .field("group", &self.group)
135            .field("listener", &self.listener.is_some())
136            .finish_non_exhaustive()
137    }
138}
krabka_client_streams/barrier/align.rs

krabka_client_streams/barrier/
align.rs
