krabka_client_streams/runtime/
app.rs

1//! `KafkaStreams`, the managed runtime handle. It owns the membership and a
2//! `StreamThread`.
3//!
4//! `start()` builds the broker I/O and joins the streams group, where the
5//! membership owns the heartbeat. It then spawns a supervisor that pumps the
6//! membership events into a `StreamThread` and polls and commits on intervals.
7//! `close()` stops the supervisor, which flushes, commits, and leaves the
8//! group.
9
10use std::{sync::Arc, time::Duration};
11
12use krabka_client_core::{
13    ClientDnsTimeout, ClientFrameMax, ConnectionDispatchQueueCapacity, DEFAULT_CLIENT_FRAME_MAX,
14    DEFAULT_CONNECTION_DISPATCH_QUEUE_CAPACITY, DEFAULT_FETCH_MIN, FetchMinBytes,
15};
16use krabka_units::prelude::*;
17use refined_type::rule::{MinMaxI64, MinMaxU128, MinMaxUsize};
18use tokio::{sync::mpsc, task::JoinHandle};
19use tokio_util::sync::CancellationToken;
20
21use crate::{
22    barrier::{BarrierAlignment, BarrierCut},
23    error::StreamsClientError,
24    membership::{
25        DEFAULT_STREAMS_JOIN_RETRY_BACKOFF, DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT,
26        DEFAULT_STREAMS_REBALANCE_TIMEOUT, StreamsEvent, StreamsJoinRetryBackoff,
27        StreamsLeaveHeartbeatTimeout, StreamsMembership, StreamsRebalanceTimeout,
28    },
29    processor::serde::Serde,
30    runtime::{
31        eos::{ProcessingGuarantee, TransactionalProducer},
32        io::{OffsetStore, RecordFetcher, RecordProducer},
33        io_broker,
34        iq::IqRequest,
35        iq_view::ReadOnlyKeyValueStore,
36        iqv2::{Query, StateQuery, StateQueryResult, dispatch::Iq2Request},
37        thread::StreamThread,
38    },
39    store::iq::StoreKind,
40    topology::BuiltTopology,
41};
42
43/// A request from a [`KafkaStreams`] handle to the supervisor, to rewind every
44/// task to a barrier cut.
45///
46/// The supervisor owns the [`StreamThread`], so a rewind travels the same way an
47/// interactive query does.
48enum BarrierRequest {
49    /// Rewind to the cut of a named epoch.
50    RestoreToEpoch {
51        epoch: i64,
52        reply: tokio::sync::oneshot::Sender<Result<BarrierCut, StreamsClientError>>,
53    },
54    /// Rewind to the complete cut with the highest epoch.
55    RestoreToLatestCut {
56        reply: tokio::sync::oneshot::Sender<Result<BarrierCut, StreamsClientError>>,
57    },
58}
59
60/// Default delay between Client Streams processing polls.
61pub const DEFAULT_STREAMS_POLL_INTERVAL: Duration = Duration::from_millis(200);
62/// Default delay between Client Streams commit attempts.
63pub const DEFAULT_STREAMS_COMMIT_INTERVAL: Duration = Duration::from_secs(5);
64/// Default capacity of each Client Streams interactive-query request queue.
65pub const DEFAULT_STREAMS_INTERACTIVE_QUERY_QUEUE_CAPACITY: usize = 64;
66/// Default Client Streams state-store record-cache budget (the JVM default).
67pub const DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES: ByteSize = mebibytes(10);
68
69/// The largest cache budget that both the public `i64` API and the target's
70/// internal `usize` cache accounting can represent.
71#[allow(clippy::cast_possible_wrap)]
72pub const MAX_STREAMS_STATE_STORE_CACHE_MAX_BYTES: i64 = if usize::BITS >= i64::BITS {
73    i64::MAX
74} else {
75    usize::MAX as i64
76};
77
78/// Target-supported Client Streams state-store record-cache budget.
79///
80/// This type holds the validated magnitude as the raw `i64` byte count that the
81/// refinement bounds are written over, so it stays `Eq`. A [`ByteSize`] stores
82/// an `f64` and cannot be `Eq`. [`size`](Self::size) puts the dimension back on
83/// at the seam.
84#[derive(Clone, Copy, Debug, Eq, PartialEq)]
85pub struct StreamsStateStoreCacheMaxBytes(i64);
86
87impl StreamsStateStoreCacheMaxBytes {
88    /// Validate a state-store cache budget.
89    ///
90    /// A zero budget disables caching.
91    ///
92    /// # Errors
93    ///
94    /// Returns an error for negative values or values that cannot be represented
95    /// by the target's internal cache accounting.
96    pub fn new(value: ByteSize) -> Result<Self, String> {
97        MinMaxI64::<0, MAX_STREAMS_STATE_STORE_CACHE_MAX_BYTES>::new(value.bytes_i64())
98            .map(|value| Self(value.into_value()))
99            .map_err(|error| format!("streams state-store cache max bytes: {error}"))
100    }
101
102    /// Return the validated budget.
103    #[must_use]
104    pub fn size(self) -> ByteSize {
105        ByteSize::from_bytes_i64(self.0)
106    }
107}
108
109impl Default for StreamsStateStoreCacheMaxBytes {
110    fn default() -> Self {
111        Self::new(DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES)
112            .expect("default streams state-store cache max bytes is valid")
113    }
114}
115
116/// Tokio-supported capacity shared by the Client Streams interactive-query queues.
117#[derive(Clone, Copy, Debug, Eq, PartialEq)]
118pub struct StreamsInteractiveQueryQueueCapacity(usize);
119
120impl StreamsInteractiveQueryQueueCapacity {
121    /// Validate an interactive-query queue capacity.
122    ///
123    /// # Errors
124    ///
125    /// Returns an error unless `value` is within Tokio's supported channel
126    /// capacity range.
127    pub fn new(value: usize) -> Result<Self, String> {
128        MinMaxUsize::<1, { tokio::sync::Semaphore::MAX_PERMITS }>::new(value)
129            .map(|value| Self(value.into_value()))
130            .map_err(|error| format!("streams interactive-query queue capacity: {error}"))
131    }
132
133    /// Return the validated capacity.
134    #[must_use]
135    pub const fn capacity(self) -> usize {
136        self.0
137    }
138}
139
140impl Default for StreamsInteractiveQueryQueueCapacity {
141    fn default() -> Self {
142        Self::new(DEFAULT_STREAMS_INTERACTIVE_QUERY_QUEUE_CAPACITY)
143            .expect("default streams interactive-query queue capacity is valid")
144    }
145}
146
147fn interactive_query_queue_capacities(
148    capacity: StreamsInteractiveQueryQueueCapacity,
149) -> [usize; 2] {
150    [capacity.capacity(); 2]
151}
152
153fn validate_positive_whole_milliseconds(field: &str, value: Duration) -> Result<u64, String> {
154    let milliseconds = MinMaxU128::<1, { u64::MAX as u128 }>::new(value.as_millis())
155        .map_err(|error| format!("{field}: {error}"))?
156        .into_value();
157    let milliseconds = u64::try_from(milliseconds).map_err(|error| format!("{field}: {error}"))?;
158    if Duration::from_millis(milliseconds) != value {
159        return Err(format!("{field} must be a whole number of milliseconds"));
160    }
161    Ok(milliseconds)
162}
163
164/// Positive, whole-millisecond Client Streams processing poll interval.
165#[derive(Clone, Copy, Debug, Eq, PartialEq)]
166pub struct StreamsPollInterval(Duration);
167
168impl StreamsPollInterval {
169    /// Validate a processing poll interval.
170    ///
171    /// # Errors
172    ///
173    /// Returns an error for zero, fractional milliseconds, or a value whose
174    /// milliseconds cannot be represented as `u64`.
175    pub fn new(value: Duration) -> Result<Self, String> {
176        validate_positive_whole_milliseconds("streams poll interval", value)?;
177        Ok(Self(value))
178    }
179
180    /// Return the validated duration.
181    #[must_use]
182    pub const fn duration(self) -> Duration {
183        self.0
184    }
185
186    /// Return the validated whole milliseconds.
187    ///
188    /// # Panics
189    ///
190    /// Panics if the validated interval no longer fits in `u64`.
191    #[must_use]
192    pub fn milliseconds(self) -> u64 {
193        u64::try_from(self.0.as_millis()).expect("validated streams poll interval fits u64")
194    }
195}
196
197impl Default for StreamsPollInterval {
198    fn default() -> Self {
199        Self::new(DEFAULT_STREAMS_POLL_INTERVAL).expect("default streams poll interval is valid")
200    }
201}
202
203/// Positive, whole-millisecond Client Streams commit interval.
204#[derive(Clone, Copy, Debug, Eq, PartialEq)]
205pub struct StreamsCommitInterval(Duration);
206
207impl StreamsCommitInterval {
208    /// Validate a commit interval.
209    ///
210    /// # Errors
211    ///
212    /// Returns an error for zero, fractional milliseconds, or a value whose
213    /// milliseconds cannot be represented as `u64`.
214    pub fn new(value: Duration) -> Result<Self, String> {
215        validate_positive_whole_milliseconds("streams commit interval", value)?;
216        Ok(Self(value))
217    }
218
219    /// Return the validated duration.
220    #[must_use]
221    pub const fn duration(self) -> Duration {
222        self.0
223    }
224
225    /// Return the validated whole milliseconds.
226    ///
227    /// # Panics
228    ///
229    /// Panics if the validated interval no longer fits in `u64`.
230    #[must_use]
231    pub fn milliseconds(self) -> u64 {
232        u64::try_from(self.0.as_millis()).expect("validated streams commit interval fits u64")
233    }
234}
235
236impl Default for StreamsCommitInterval {
237    fn default() -> Self {
238        Self::new(DEFAULT_STREAMS_COMMIT_INTERVAL)
239            .expect("default streams commit interval is valid")
240    }
241}
242
243fn validate_runtime_configuration(
244    poll_interval: Duration,
245    commit_interval: Duration,
246    rebalance_timeout: Duration,
247    join_retry_backoff: Duration,
248    leave_heartbeat_timeout: Duration,
249    cache_max_bytes: ByteSize,
250) -> Result<
251    (
252        StreamsPollInterval,
253        StreamsCommitInterval,
254        StreamsRebalanceTimeout,
255        StreamsJoinRetryBackoff,
256        StreamsLeaveHeartbeatTimeout,
257        StreamsStateStoreCacheMaxBytes,
258    ),
259    StreamsClientError,
260> {
261    let poll_interval =
262        StreamsPollInterval::new(poll_interval).map_err(StreamsClientError::Runtime)?;
263    let commit_interval =
264        StreamsCommitInterval::new(commit_interval).map_err(StreamsClientError::Runtime)?;
265    let rebalance_timeout =
266        StreamsRebalanceTimeout::new(rebalance_timeout).map_err(StreamsClientError::Runtime)?;
267    let join_retry_backoff =
268        StreamsJoinRetryBackoff::new(join_retry_backoff).map_err(StreamsClientError::Runtime)?;
269    let leave_heartbeat_timeout = StreamsLeaveHeartbeatTimeout::new(leave_heartbeat_timeout)
270        .map_err(StreamsClientError::Runtime)?;
271    let cache_max_bytes = StreamsStateStoreCacheMaxBytes::new(cache_max_bytes)
272        .map_err(StreamsClientError::Runtime)?;
273    Ok((
274        poll_interval,
275        commit_interval,
276        rebalance_timeout,
277        join_retry_backoff,
278        leave_heartbeat_timeout,
279        cache_max_bytes,
280    ))
281}
282
283/// A managed Kafka Streams runtime.
284///
285/// It joins a streams group, runs the assigned tasks in the at-least-once order
286/// fetch → process → produce → commit, and reacts to rebalances.
287pub struct KafkaStreams {
288    member_id: String,
289    shutdown: CancellationToken,
290    handle: JoinHandle<Result<(), StreamsClientError>>,
291    /// Channel to the supervisor for interactive queries. The `KafkaStreams` IQ
292    /// accessors read it.
293    iq_tx: mpsc::Sender<IqRequest>,
294    /// Channel to the supervisor for `IQv2` queries. It is separate from the v1
295    /// `iq_tx`.
296    iq2_tx: mpsc::Sender<Iq2Request>,
297    /// Channel to the supervisor for barrier rewinds.
298    barrier_tx: mpsc::Sender<BarrierRequest>,
299}
300
301#[bon::bon]
302impl KafkaStreams {
303    #[builder(start_fn = builder, finish_fn = build)]
304    // one-shot constructor: broker I/O setup,
305    // membership join, and the supervisor select-loop (now two IQ channels).
306    #[tracing::instrument(
307        name = "streams.app.start",
308        level = "info",
309        skip_all,
310        fields(
311            application_id = %application_id,
312            processing_guarantee = ?processing_guarantee,
313            member_id = tracing::field::Empty,
314        ),
315        err,
316    )]
317    /// # Errors
318    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
319    #[allow(clippy::similar_names)]
320    pub async fn start(
321        #[builder(into)] bootstrap: String,
322        #[builder(into)] application_id: String,
323        topology: BuiltTopology,
324        #[builder(default = DEFAULT_STREAMS_POLL_INTERVAL)] poll_interval: Duration,
325        #[builder(default = DEFAULT_STREAMS_COMMIT_INTERVAL)] commit_interval: Duration,
326        #[builder(default = DEFAULT_STREAMS_REBALANCE_TIMEOUT)] rebalance_timeout: Duration,
327        #[builder(default = DEFAULT_STREAMS_JOIN_RETRY_BACKOFF)] join_retry_backoff: Duration,
328        #[builder(default = DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT)]
329        leave_heartbeat_timeout: Duration,
330        #[builder(default)] store_backend: crate::store::backend::StoreBackend,
331        #[builder(default)] processing_guarantee: crate::runtime::eos::ProcessingGuarantee,
332        /// Deadline for each Kafka broker DNS lookup owned by this process.
333        #[builder(default)]
334        broker_dns_timeout: ClientDnsTimeout,
335        #[builder(default = DEFAULT_CONNECTION_DISPATCH_QUEUE_CAPACITY)]
336        client_dispatch_queue_capacity: usize,
337        #[builder(default = DEFAULT_CLIENT_FRAME_MAX)] client_frame_max: ByteSize,
338        #[builder(default = DEFAULT_FETCH_MIN)] fetch_min: ByteSize,
339        /// Record-cache budget, which is the JVM `statestore.cache.max.bytes`.
340        /// `0` turns caching off. `instantiate` threads it onto each task
341        /// graph.
342        #[builder(default = DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES)]
343        cache_max_bytes: ByteSize,
344        /// Capacity shared by the v1 and v2 interactive-query request queues.
345        #[builder(default)]
346        interactive_query_queue_capacity: StreamsInteractiveQueryQueueCapacity,
347        /// The barrier group to align on, and where each cut's snapshot lives.
348        /// Without one the runtime holds no record back and takes no snapshot.
349        barrier: Option<BarrierAlignment>,
350    ) -> Result<Self, StreamsClientError> {
351        let (
352            poll_interval,
353            commit_interval,
354            rebalance_timeout,
355            join_retry_backoff,
356            leave_heartbeat_timeout,
357            cache_max_bytes,
358        ) = validate_runtime_configuration(
359            poll_interval,
360            commit_interval,
361            rebalance_timeout,
362            join_retry_backoff,
363            leave_heartbeat_timeout,
364            cache_max_bytes,
365        )?;
366        let cache_max_bytes = cache_max_bytes.size();
367        let client_dispatch_queue_capacity =
368            ConnectionDispatchQueueCapacity::new(client_dispatch_queue_capacity)
369                .map_err(StreamsClientError::Runtime)?;
370        let client_frame_max =
371            ClientFrameMax::try_from(client_frame_max).map_err(StreamsClientError::Runtime)?;
372        let fetch_min = FetchMinBytes::try_from(fetch_min).map_err(StreamsClientError::Runtime)?;
373        let client_resource_policy = io_broker::ClientResourcePolicy {
374            dispatch_queue_capacity: client_dispatch_queue_capacity,
375            frame_max: client_frame_max,
376            fetch_min,
377        };
378        let built = Arc::new(topology);
379
380        // Broker I/O. Under EOS-v2 the producer is transactional: the SAME object
381        // is both the task `RecordProducer` (for `send`) and the thread's
382        // `TransactionalProducer` (for begin/send_offsets/commit). Under ALO `txn`
383        // is `None`.
384        let fetcher: Arc<dyn RecordFetcher>;
385        let producer: Arc<dyn RecordProducer>;
386        let store: Arc<io_broker::BrokerOffsetStore>;
387        let txn: Option<Arc<dyn TransactionalProducer>>;
388        match processing_guarantee {
389            ProcessingGuarantee::AtLeastOnce => {
390                let (f, p, s) = io_broker::build(
391                    &bootstrap,
392                    &application_id,
393                    &application_id,
394                    broker_dns_timeout,
395                    client_resource_policy,
396                )
397                .await?;
398                fetcher = Arc::new(f);
399                producer = p;
400                store = s;
401                txn = None;
402            }
403            ProcessingGuarantee::ExactlyOnceV2 => {
404                let txn_id = crate::runtime::eos::transactional_id(&application_id, 0);
405                let (f, txn_producer, s) = io_broker::build_eos(
406                    &bootstrap,
407                    &application_id,
408                    &application_id,
409                    &txn_id,
410                    broker_dns_timeout,
411                    client_resource_policy,
412                )
413                .await?;
414                fetcher = Arc::new(f);
415                // Two trait-object views of the one transactional producer.
416                producer = Arc::clone(&txn_producer) as Arc<dyn RecordProducer>;
417                txn = Some(txn_producer as Arc<dyn TransactionalProducer>);
418                store = s;
419            }
420        }
421
422        let offset_store = Arc::clone(&store);
423        let store: Arc<dyn OffsetStore> = store;
424
425        // Join the streams group (membership owns the heartbeat loop).
426        let mut membership = StreamsMembership::builder()
427            .bootstrap(bootstrap.clone())
428            .group_id(application_id.clone())
429            .topology(Arc::clone(&built))
430            .broker_dns_timeout(broker_dns_timeout)
431            .dispatch_queue_capacity(client_dispatch_queue_capacity)
432            .frame_max(client_frame_max)
433            .rebalance_timeout(rebalance_timeout.duration())
434            .join_retry_backoff(join_retry_backoff.duration())
435            .leave_heartbeat_timeout(leave_heartbeat_timeout.duration())
436            .build()
437            .await?;
438        let member_id = membership.member_id().to_string();
439        tracing::Span::current().record("member_id", tracing::field::display(&member_id));
440
441        // Supervisor: pump membership events into a StreamThread + poll/commit.
442        let shutdown = CancellationToken::new();
443        let sd = shutdown.clone();
444        let topo_for_thread = Arc::clone(&built);
445        let fetcher_for_thread = Arc::clone(&fetcher);
446        let [iq_capacity, iq2_capacity] =
447            interactive_query_queue_capacities(interactive_query_queue_capacity);
448        let (iq_tx, mut iq_rx) = mpsc::channel::<IqRequest>(iq_capacity);
449        let (iq2_tx, mut iq2_rx) = mpsc::channel::<Iq2Request>(iq2_capacity);
450        // One in flight is enough: a rewind is an operator action, not a query.
451        let (barrier_tx, mut barrier_rx) = mpsc::channel::<BarrierRequest>(1);
452        let is_eos = processing_guarantee == ProcessingGuarantee::ExactlyOnceV2;
453        let handle = tokio::spawn(async move {
454            let mut thread = StreamThread::new(
455                fetcher_for_thread,
456                store_backend,
457                application_id,
458                cache_max_bytes,
459            );
460            if let Some(alignment) = barrier {
461                thread = thread.with_barrier(alignment);
462            }
463            let mut poll = tokio::time::interval(poll_interval.duration());
464            let mut commit = tokio::time::interval(commit_interval.duration());
465            commit.tick().await;
466            let tracker = membership.tracker();
467            loop {
468                tokio::select! {
469                    () = sd.cancelled() => {
470                        // EOS close aborts any in-flight txn (meta unused); ALO commits.
471                        let meta = membership.group_metadata().await;
472                        offset_store.update_group_metadata(meta.clone()).await;
473                        let closed = thread.close_all(Some(&meta)).await;
474                        let left = membership.close().await;
475                        closed?;
476                        left?;
477                        break;
478                    }
479                    ev = membership.next_event() => match ev {
480                        Ok(StreamsEvent::Assigned(a)) => {
481                            offset_store.update_group_metadata(membership.group_metadata().await).await;
482                            if let Err(e) = thread
483                                .apply_assignment(
484                                    &a,
485                                    &topo_for_thread,
486                                    &producer,
487                                    &store,
488                                    processing_guarantee,
489                                    txn.clone(),
490                                )
491                                .await
492                            {
493                                tracing::warn!(error = %e, "apply_assignment failed");
494                            }
495                        }
496                        Ok(StreamsEvent::Fenced) => { let _ = thread.close_all(None).await; }
497                        Ok(StreamsEvent::NotReady(_)) => {}
498                        Err(e) => { tracing::warn!(error = %e, "membership event stream ended"); break; }
499                    },
500                    _ = poll.tick() => {
501                        if let Err(e) = thread.poll_all(&*fetcher, &tracker).await {
502                            tracing::warn!(error = %e, "poll_all failed");
503                        }
504                    }
505                    _ = commit.tick() => {
506                        // EOS commit folds offsets into the txn — needs the live
507                        // streams group metadata; ALO puts it on OffsetCommit.
508                        let meta = membership.group_metadata().await;
509                        offset_store.update_group_metadata(meta.clone()).await;
510                        if let Err(e) = thread.commit_all(is_eos.then_some(&meta)).await {
511                            tracing::warn!(error = %e, "commit_all failed");
512                        }
513                    }
514                    Some(req) = iq_rx.recv() => {
515                        thread.serve_iq(req).await;
516                    }
517                    Some(req) = iq2_rx.recv() => {
518                        thread.serve_iq2(req).await;
519                    }
520                    Some(req) = barrier_rx.recv() => {
521                        match req {
522                            BarrierRequest::RestoreToEpoch { epoch, reply } => {
523                                let _ = reply.send(thread.restore_to_epoch(epoch).await);
524                            }
525                            BarrierRequest::RestoreToLatestCut { reply } => {
526                                let _ = reply.send(thread.restore_to_latest_cut().await);
527                            }
528                        }
529                    }
530                }
531            }
532            Ok(())
533        });
534
535        Ok(Self {
536            member_id,
537            shutdown,
538            handle,
539            iq_tx,
540            iq2_tx,
541            barrier_tx,
542        })
543    }
544}
545
546impl KafkaStreams {
547    /// The client-generated streams member id.
548    #[must_use]
549    pub fn member_id(&self) -> &str {
550        &self.member_id
551    }
552
553    /// A read-only view of the local `KeyValue` state store `name`, for
554    /// interactive queries.
555    ///
556    /// It returns an error when this instance holds no assignment for the store,
557    /// or when the store is of a different kind.
558    /// # Errors
559    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
560    pub async fn key_value_store<K, V>(
561        &self,
562        name: impl Into<String>,
563        key_serde: impl Serde<K> + 'static,
564        value_serde: impl Serde<V> + 'static,
565    ) -> Result<ReadOnlyKeyValueStore<K, V>, StreamsClientError> {
566        let view = ReadOnlyKeyValueStore {
567            tx: self.iq_tx.clone(),
568            store: name.into(),
569            key_serde: Box::new(key_serde),
570            value_serde: Box::new(value_serde),
571        };
572        crate::runtime::iq_view::validate(&view.tx, &view.store, StoreKind::KeyValue).await?;
573        Ok(view)
574    }
575
576    /// A read-only view of the local `Window` state store `name`.
577    /// # Errors
578    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
579    pub async fn window_store<K, V>(
580        &self,
581        name: impl Into<String>,
582        key_serde: impl Serde<K> + 'static,
583        value_serde: impl Serde<V> + 'static,
584    ) -> Result<crate::runtime::iq_view::ReadOnlyWindowStore<K, V>, StreamsClientError> {
585        let view = crate::runtime::iq_view::ReadOnlyWindowStore {
586            tx: self.iq_tx.clone(),
587            store: name.into(),
588            key_serde: Box::new(key_serde),
589            value_serde: Box::new(value_serde),
590        };
591        crate::runtime::iq_view::validate(&view.tx, &view.store, StoreKind::Window).await?;
592        Ok(view)
593    }
594
595    /// A read-only view of the local `Session` state store `name`.
596    /// # Errors
597    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
598    pub async fn session_store<K, V>(
599        &self,
600        name: impl Into<String>,
601        key_serde: impl Serde<K> + 'static,
602        value_serde: impl Serde<V> + 'static,
603    ) -> Result<crate::runtime::iq_view::ReadOnlySessionStore<K, V>, StreamsClientError> {
604        let view = crate::runtime::iq_view::ReadOnlySessionStore {
605            tx: self.iq_tx.clone(),
606            store: name.into(),
607            key_serde: Box::new(key_serde),
608            value_serde: Box::new(value_serde),
609        };
610        crate::runtime::iq_view::validate(&view.tx, &view.store, StoreKind::Session).await?;
611        Ok(view)
612    }
613
614    /// Run an `IQv2` query against the locally assigned partitions and return
615    /// one `QueryResult` per partition. The method takes no serdes, because the
616    /// store supplies its own.
617    pub async fn query<Q: Query>(&self, req: StateQuery<Q>) -> StateQueryResult<Q::Result> {
618        use crate::runtime::iqv2::dispatch::{Iq2Request, assemble};
619
620        let kind = req.query.store_kind();
621        let (reply, rx) = tokio::sync::oneshot::channel();
622        let iq2 = Iq2Request {
623            store: req.store,
624            kind,
625            query: req.query.lower(),
626            partitions: req.partitions,
627            bound: req.bound,
628            require_active: req.require_active,
629            reply,
630        };
631        if self.iq2_tx.send(iq2).await.is_err() {
632            return StateQueryResult::new(std::collections::BTreeMap::new());
633        }
634        match rx.await {
635            Ok(outcome) => assemble::<Q::Result>(outcome),
636            Err(_) => StateQueryResult::new(std::collections::BTreeMap::new()),
637        }
638    }
639
640    /// Rewinds every active task to the cut of `epoch`.
641    ///
642    /// Each task's stores become the snapshot taken at that cut, and every
643    /// source partition the cut names seeks to its marker offset. Processing
644    /// then replays from the cut.
645    ///
646    /// # Errors
647    ///
648    /// Returns an error when the app names no barrier group, when the group has
649    /// no complete cut at `epoch`, when a task has no snapshot at that epoch, or
650    /// when the supervisor has stopped.
651    pub async fn restore_to_epoch(&self, epoch: i64) -> Result<BarrierCut, StreamsClientError> {
652        let (reply, rx) = tokio::sync::oneshot::channel();
653        self.send_barrier(BarrierRequest::RestoreToEpoch { epoch, reply }, rx)
654            .await
655    }
656
657    /// Rewinds every active task to the complete cut with the highest epoch, as
658    /// [`restore_to_epoch`](Self::restore_to_epoch) does for a named epoch.
659    ///
660    /// # Errors
661    ///
662    /// Returns an error when the app names no barrier group, when the group has
663    /// no complete cut, when a task has no snapshot at that cut, or when the
664    /// supervisor has stopped.
665    pub async fn restore_to_latest_cut(&self) -> Result<BarrierCut, StreamsClientError> {
666        let (reply, rx) = tokio::sync::oneshot::channel();
667        self.send_barrier(BarrierRequest::RestoreToLatestCut { reply }, rx)
668            .await
669    }
670
671    /// Sends one rewind to the supervisor and waits for its answer.
672    async fn send_barrier(
673        &self,
674        request: BarrierRequest,
675        rx: tokio::sync::oneshot::Receiver<Result<BarrierCut, StreamsClientError>>,
676    ) -> Result<BarrierCut, StreamsClientError> {
677        let stopped = || StreamsClientError::Runtime("streams runtime has stopped".to_owned());
678        self.barrier_tx
679            .send(request)
680            .await
681            .map_err(|_error| stopped())?;
682        rx.await.map_err(|_error| stopped())?
683    }
684
685    /// Stop processing, commit, and leave the group.
686    #[tracing::instrument(
687        name = "streams.app.close",
688        level = "info",
689        skip_all,
690        fields(member_id = %self.member_id),
691        err,
692    )]
693    /// # Errors
694    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
695    pub async fn close(self) -> Result<(), StreamsClientError> {
696        self.shutdown.cancel();
697        self.handle.await.map_err(|error| {
698            StreamsClientError::Runtime(format!("streams supervisor failed: {error}"))
699        })?
700    }
701
702    /// Stop the runtime without committing, aborting transactions, or leaving
703    /// the group. Tests use this to model an abruptly killed process.
704    pub async fn crash(self) {
705        self.handle.abort();
706        let _ = self.handle.await;
707    }
708}
709
710#[cfg(test)]
711mod tests {
712    use std::time::Duration;
713
714    use assert2::check;
715    use krabka_units::prelude::*;
716
717    use super::{
718        DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES, KafkaStreams,
719        MAX_STREAMS_STATE_STORE_CACHE_MAX_BYTES, StreamsCommitInterval,
720        StreamsInteractiveQueryQueueCapacity, StreamsPollInterval, StreamsStateStoreCacheMaxBytes,
721        interactive_query_queue_capacities, validate_runtime_configuration,
722    };
723    use crate::{
724        membership::{DEFAULT_STREAMS_JOIN_RETRY_BACKOFF, DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT},
725        topology::Topology,
726    };
727
728    #[test]
729    fn interactive_query_queue_capacity_uses_default_and_valid_override() {
730        let default = StreamsInteractiveQueryQueueCapacity::default();
731        assert_eq!(default.capacity(), 64);
732
733        let capacity =
734            StreamsInteractiveQueryQueueCapacity::new(37).expect("positive queue capacity");
735        assert_eq!(capacity.capacity(), 37);
736    }
737
738    #[test]
739    fn interactive_query_queue_capacity_rejects_zero() {
740        let error = StreamsInteractiveQueryQueueCapacity::new(0).expect_err("zero queue capacity");
741        assert2::assert!(error.contains("streams interactive-query queue capacity"));
742    }
743
744    #[test]
745    fn interactive_query_queue_capacity_matches_tokio_boundaries() {
746        let maximum =
747            StreamsInteractiveQueryQueueCapacity::new(tokio::sync::Semaphore::MAX_PERMITS)
748                .expect("Tokio maximum queue capacity");
749        assert_eq!(maximum.capacity(), tokio::sync::Semaphore::MAX_PERMITS);
750
751        StreamsInteractiveQueryQueueCapacity::new(tokio::sync::Semaphore::MAX_PERMITS + 1)
752            .expect_err("capacity above Tokio maximum");
753    }
754
755    #[test]
756    fn interactive_query_queues_share_the_configured_capacity() {
757        let capacity =
758            StreamsInteractiveQueryQueueCapacity::new(37).expect("positive queue capacity");
759        assert_eq!(interactive_query_queue_capacities(capacity), [37, 37]);
760    }
761
762    #[test]
763    fn state_store_cache_max_bytes_uses_default_and_valid_overrides() {
764        let default = StreamsStateStoreCacheMaxBytes::default();
765        check!(default.size() == mebibytes(10));
766        check!(
767            StreamsStateStoreCacheMaxBytes::new(ByteSize::ZERO)
768                .expect("zero disables caching")
769                .size()
770                == ByteSize::ZERO
771        );
772        check!(
773            StreamsStateStoreCacheMaxBytes::new(bytes(37))
774                .expect("positive cache budget")
775                .size()
776                == bytes(37)
777        );
778    }
779
780    #[test]
781    fn state_store_cache_max_bytes_rejects_negative_values() {
782        let error = StreamsStateStoreCacheMaxBytes::new(ByteSize::from_bytes_i64(-1))
783            .expect_err("negative cache budget");
784        check!(error.contains("streams state-store cache max bytes"));
785    }
786
787    #[test]
788    fn state_store_cache_max_bytes_matches_target_boundaries() {
789        let maximum = StreamsStateStoreCacheMaxBytes::new(ByteSize::from_bytes_i64(
790            MAX_STREAMS_STATE_STORE_CACHE_MAX_BYTES,
791        ))
792        .expect("target-supported maximum");
793        check!(maximum.size() == ByteSize::from_bytes_i64(MAX_STREAMS_STATE_STORE_CACHE_MAX_BYTES));
794
795        if let Some(too_large) = MAX_STREAMS_STATE_STORE_CACHE_MAX_BYTES.checked_add(1) {
796            StreamsStateStoreCacheMaxBytes::new(ByteSize::from_bytes_i64(too_large))
797                .expect_err("value above target-supported maximum");
798        }
799    }
800
801    #[test]
802    fn runtime_intervals_use_typed_defaults_and_valid_overrides() {
803        let poll = StreamsPollInterval::default();
804        let commit = StreamsCommitInterval::default();
805        assert2::assert!(poll.milliseconds() == 200);
806        assert2::assert!(commit.milliseconds() == 5_000);
807
808        let poll = StreamsPollInterval::new(Duration::from_millis(37))
809            .expect("positive whole milliseconds");
810        let commit = StreamsCommitInterval::new(Duration::from_millis(41))
811            .expect("positive whole milliseconds");
812        assert2::assert!(poll.duration() == Duration::from_millis(37));
813        assert2::assert!(commit.duration() == Duration::from_millis(41));
814    }
815
816    #[test]
817    fn runtime_intervals_reject_zero_and_fractional_milliseconds() {
818        assert2::assert!(StreamsPollInterval::new(Duration::ZERO).is_err());
819        assert2::assert!(StreamsCommitInterval::new(Duration::ZERO).is_err());
820        assert2::assert!(
821            StreamsPollInterval::new(Duration::from_millis(1) + Duration::from_nanos(1)).is_err()
822        );
823        assert2::assert!(
824            StreamsCommitInterval::new(Duration::from_millis(1) + Duration::from_nanos(1)).is_err()
825        );
826    }
827
828    #[test]
829    fn low_level_runtime_validation_names_the_invalid_field() {
830        let poll_error = validate_runtime_configuration(
831            Duration::ZERO,
832            Duration::from_secs(5),
833            Duration::from_secs(30),
834            DEFAULT_STREAMS_JOIN_RETRY_BACKOFF,
835            DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT,
836            DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES,
837        )
838        .expect_err("zero poll interval");
839        assert2::assert!(poll_error.to_string().contains("streams poll interval"));
840
841        let commit_error = validate_runtime_configuration(
842            Duration::from_millis(200),
843            Duration::ZERO,
844            Duration::from_secs(30),
845            DEFAULT_STREAMS_JOIN_RETRY_BACKOFF,
846            DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT,
847            DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES,
848        )
849        .expect_err("zero commit interval");
850        assert2::assert!(commit_error.to_string().contains("streams commit interval"));
851
852        let rebalance_error = validate_runtime_configuration(
853            Duration::from_millis(200),
854            Duration::from_secs(5),
855            Duration::from_millis(u64::try_from(i32::MAX).expect("i32 max fits u64") + 1),
856            DEFAULT_STREAMS_JOIN_RETRY_BACKOFF,
857            DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT,
858            DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES,
859        )
860        .expect_err("rebalance timeout outside Kafka wire range");
861        assert2::assert!(
862            rebalance_error
863                .to_string()
864                .contains("streams rebalance timeout")
865        );
866
867        let join_retry_error = validate_runtime_configuration(
868            Duration::from_millis(200),
869            Duration::from_secs(5),
870            Duration::from_secs(30),
871            Duration::ZERO,
872            DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT,
873            DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES,
874        )
875        .expect_err("zero join retry backoff");
876        assert2::assert!(
877            join_retry_error
878                .to_string()
879                .contains("streams join retry backoff")
880        );
881
882        let leave_error = validate_runtime_configuration(
883            Duration::from_millis(200),
884            Duration::from_secs(5),
885            Duration::from_secs(30),
886            DEFAULT_STREAMS_JOIN_RETRY_BACKOFF,
887            Duration::ZERO,
888            DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES,
889        )
890        .expect_err("zero leave heartbeat timeout");
891        assert2::assert!(
892            leave_error
893                .to_string()
894                .contains("streams leave heartbeat timeout")
895        );
896
897        let cache_error = validate_runtime_configuration(
898            Duration::from_millis(200),
899            Duration::from_secs(5),
900            Duration::from_secs(30),
901            DEFAULT_STREAMS_JOIN_RETRY_BACKOFF,
902            DEFAULT_STREAMS_LEAVE_HEARTBEAT_TIMEOUT,
903            ByteSize::from_bytes_i64(-1),
904        )
905        .expect_err("negative cache budget");
906        check!(
907            cache_error
908                .to_string()
909                .contains("streams state-store cache max bytes")
910        );
911    }
912
913    #[tokio::test]
914    async fn invalid_leave_heartbeat_timeout_fails_before_broker_lookup() {
915        let mut topology = Topology::new();
916        let source = topology.add_source::<String, String>("source", ["input"]);
917        topology.add_sink("sink", "output", [&source]);
918        let topology = topology.build("leave-validation").expect("topology");
919
920        let error = KafkaStreams::builder()
921            .bootstrap("invalid.invalid:9092")
922            .application_id("leave-validation")
923            .topology(topology)
924            .leave_heartbeat_timeout(Duration::ZERO)
925            .build()
926            .await
927            .err()
928            .expect("invalid configuration");
929
930        assert2::assert!(
931            error
932                .to_string()
933                .contains("streams leave heartbeat timeout")
934        );
935    }
936
937    #[tokio::test]
938    async fn invalid_join_retry_backoff_fails_before_broker_lookup() {
939        let mut topology = Topology::new();
940        let source = topology.add_source::<String, String>("source", ["input"]);
941        topology.add_sink("sink", "output", [&source]);
942        let topology = topology.build("join-retry-validation").expect("topology");
943
944        let error = KafkaStreams::builder()
945            .bootstrap("invalid.invalid:9092")
946            .application_id("join-retry-validation")
947            .topology(topology)
948            .join_retry_backoff(Duration::ZERO)
949            .build()
950            .await
951            .err()
952            .expect("invalid configuration");
953
954        assert2::assert!(error.to_string().contains("streams join retry backoff"));
955    }
956
957    #[tokio::test]
958    async fn invalid_state_store_cache_budget_fails_before_broker_lookup() {
959        let mut topology = Topology::new();
960        let source = topology.add_source::<String, String>("source", ["input"]);
961        topology.add_sink("sink", "output", [&source]);
962        let topology = topology.build("cache-budget-validation").expect("topology");
963
964        let error = KafkaStreams::builder()
965            .bootstrap("invalid.invalid:9092")
966            .application_id("cache-budget-validation")
967            .topology(topology)
968            .cache_max_bytes(ByteSize::from_bytes_i64(-1))
969            .build()
970            .await
971            .err()
972            .expect("invalid configuration");
973
974        assert2::assert!(
975            error
976                .to_string()
977                .contains("streams state-store cache max bytes")
978        );
979    }
980
981    #[tokio::test]
982    async fn invalid_client_resource_policy_fails_before_broker_lookup() {
983        let mut topology = Topology::new();
984        let source = topology.add_source::<String, String>("source", ["input"]);
985        topology.add_sink("sink", "output", [&source]);
986        let topology = topology
987            .build("client-policy-validation")
988            .expect("topology");
989
990        let error = KafkaStreams::builder()
991            .bootstrap("invalid.invalid:9092")
992            .application_id("client-policy-validation")
993            .topology(topology)
994            .client_dispatch_queue_capacity(0)
995            .build()
996            .await
997            .err()
998            .expect("invalid client policy");
999
1000        assert2::assert!(error.to_string().contains("client dispatch queue capacity"));
1001    }
1002}
krabka_client_streams/runtime/app.rs

krabka_client_streams/runtime/
app.rs
