Skip to main content

krabka_client_streams/
streams_app.rs

1//! [`StreamsApp`] is one component that owns the schema-registry lifecycle and
2//! the managed [`KafkaStreams`] runtime. An application therefore does not
3//! hand-wire the cache, `set_default_registry`, the pre-warm, and
4//! `KafkaStreams::builder()`.
5//!
6//! Build it **first**, because that installs the process default registry. Then
7//! build a topology against it and `run` the topology. The registry must be
8//! installed before the code builds the topology, because the default Avro,
9//! Protobuf, and JSON serdes read it when the DSL or
10//! [`Topology`](crate::Topology) builds them.
11//!
12//! ```no_run
13//! use apache_avro::AvroSchema;
14//! use krabka_client_streams::{DefaultSerde, SchemaSerde, StreamsApp};
15//! use krabka_schema_serde::format::avro::AvroSerde;
16//! use serde::{Deserialize, Serialize};
17//!
18//! #[derive(Clone, Serialize, Deserialize, AvroSchema)]
19//! struct Order {
20//!     id: String,
21//!     total: f64,
22//! }
23//! impl DefaultSerde for Order {
24//!     type Serde = SchemaSerde<Order, AvroSerde<Order>>;
25//! }
26//!
27//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
28//! let app = StreamsApp::builder()
29//!     .bootstrap("127.0.0.1:9092")
30//!     .application_id("orders")
31//!     .schema_registry("http://127.0.0.1:8081")
32//!     .build();
33//!
34//! let topology = app.streams_builder();
35//! topology
36//!     .stream::<String, Order>(["orders"])
37//!     .map_values(|o: &Order| Order {
38//!         id: o.id.clone(),
39//!         total: o.total * 2.0,
40//!     })
41//!     .to("orders-doubled");
42//!
43//! let streams = app.run(topology).await?;
44//! streams.close().await?;
45//! # Ok(()) }
46//! ```
47
48use std::sync::Arc;
49
50use krabka_schema_serde::{
51    RegistryClient,
52    cache::{CacheConfig, SchemaCache},
53    set_default_registry,
54};
55use krabka_units::prelude::*;
56
57use crate::{
58    DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES, StreamsInteractiveQueryQueueCapacity,
59    StreamsJoinRetryBackoff, StreamsLeaveHeartbeatTimeout, StreamsRebalanceTimeout,
60    barrier::BarrierAlignment,
61    dsl::StreamsBuilder,
62    error::StreamsClientError,
63    runtime::{KafkaStreams, StreamsCommitInterval, StreamsPollInterval, eos::ProcessingGuarantee},
64    store::StoreBackend,
65    topology::BuiltTopology,
66};
67
68/// Owns the schema-registry lifecycle and the [`KafkaStreams`] wiring for a
69/// streams application.
70///
71/// The lifecycle covers the cache, the process default, and the pre-warm. Build
72/// this type with [`StreamsApp::builder`].
73pub struct StreamsApp {
74    bootstrap: String,
75    application_id: String,
76    cache: Arc<SchemaCache>,
77    store_backend: StoreBackend,
78    processing_guarantee: ProcessingGuarantee,
79    poll_interval: StreamsPollInterval,
80    commit_interval: StreamsCommitInterval,
81    rebalance_timeout: StreamsRebalanceTimeout,
82    join_retry_backoff: StreamsJoinRetryBackoff,
83    leave_heartbeat_timeout: StreamsLeaveHeartbeatTimeout,
84    broker_dns_timeout: krabka_client_core::ClientDnsTimeout,
85    client_dispatch_queue_capacity: krabka_client_core::ConnectionDispatchQueueCapacity,
86    client_frame_max: krabka_client_core::ClientFrameMax,
87    fetch_min: krabka_client_core::FetchMinBytes,
88    cache_max_bytes: ByteSize,
89    interactive_query_queue_capacity: StreamsInteractiveQueryQueueCapacity,
90    barrier: Option<BarrierAlignment>,
91}
92
93#[bon::bon]
94impl StreamsApp {
95    /// Configure the app and **install the process default schema registry**.
96    ///
97    /// `build` installs the registry as a side effect, so the registry is ready
98    /// before the code builds the topology and its default serdes.
99    #[must_use]
100    #[builder(start_fn = builder, finish_fn = build)]
101    pub fn new(
102        #[builder(into)] bootstrap: String,
103        #[builder(into)] application_id: String,
104        /// Schema Registry base URL, e.g. `http://localhost:8081`.
105        #[builder(into)]
106        schema_registry: String,
107        /// Registry behavior, either auto-register or lookup. Default:
108        /// auto-register.
109        cache_config: Option<CacheConfig>,
110        #[builder(default)] store_backend: StoreBackend,
111        #[builder(default)] processing_guarantee: ProcessingGuarantee,
112        /// Delay between Client Streams processing polls.
113        #[builder(default)]
114        poll_interval: StreamsPollInterval,
115        /// Delay between Client Streams commit attempts.
116        #[builder(default)]
117        commit_interval: StreamsCommitInterval,
118        /// Timeout advertised for completing a Client Streams rebalance.
119        #[builder(default)]
120        rebalance_timeout: StreamsRebalanceTimeout,
121        /// Delay between Client Streams initial join retries.
122        #[builder(default)]
123        join_retry_backoff: StreamsJoinRetryBackoff,
124        /// Deadline for the final Client Streams leave heartbeat during shutdown.
125        #[builder(default)]
126        leave_heartbeat_timeout: StreamsLeaveHeartbeatTimeout,
127        /// Deadline for each Kafka broker DNS lookup owned by this process.
128        #[builder(default)]
129        broker_dns_timeout: krabka_client_core::ClientDnsTimeout,
130        /// Capacity shared by every outbound Kafka connection.
131        #[builder(default)]
132        client_dispatch_queue_capacity: krabka_client_core::ConnectionDispatchQueueCapacity,
133        /// Maximum frame size shared by every outbound Kafka connection.
134        #[builder(default)]
135        client_frame_max: krabka_client_core::ClientFrameMax,
136        /// Minimum bytes requested by broker fetches.
137        #[builder(default)]
138        fetch_min: krabka_client_core::FetchMinBytes,
139        /// Record-cache budget, which is the JVM `statestore.cache.max.bytes`.
140        /// `0` turns caching off. Default: 10 MiB, which matches the JVM
141        /// default.
142        #[builder(default = DEFAULT_STREAMS_STATE_STORE_CACHE_MAX_BYTES)]
143        cache_max_bytes: ByteSize,
144        /// Capacity shared by the v1 and v2 interactive-query request queues.
145        #[builder(default)]
146        interactive_query_queue_capacity: StreamsInteractiveQueryQueueCapacity,
147        /// The barrier group to align on, and where each cut's snapshot lives.
148        /// Without one the app holds no record back and takes no snapshot.
149        barrier: Option<BarrierAlignment>,
150    ) -> Self {
151        let cache = SchemaCache::new(
152            RegistryClient::new(schema_registry),
153            cache_config.unwrap_or_default(),
154        );
155        set_default_registry(cache.clone());
156        Self {
157            bootstrap,
158            application_id,
159            cache,
160            store_backend,
161            processing_guarantee,
162            poll_interval,
163            commit_interval,
164            rebalance_timeout,
165            join_retry_backoff,
166            leave_heartbeat_timeout,
167            broker_dns_timeout,
168            client_dispatch_queue_capacity,
169            client_frame_max,
170            fetch_min,
171            cache_max_bytes,
172            interactive_query_queue_capacity,
173            barrier,
174        }
175    }
176}
177
178impl StreamsApp {
179    /// A fresh DSL builder. The schema registry is already installed.
180    #[must_use]
181    // cargo-mutants: trivial builder accessor.
182    #[cfg_attr(test, mutants::skip)]
183    pub fn streams_builder(&self) -> StreamsBuilder {
184        StreamsBuilder::new()
185    }
186
187    /// The configured application id. Pass it to [`Topology::build`] when you
188    /// wire a Processor-API topology by hand and feed that topology to
189    /// [`run_built`](Self::run_built).
190    ///
191    /// [`Topology::build`]: crate::Topology::build
192    #[must_use]
193    pub fn application_id(&self) -> &str {
194        &self.application_id
195    }
196
197    /// Lower the DSL topology, pre-warm its schema subjects, and start the app.
198    #[tracing::instrument(
199        name = "streams.app.run",
200        level = "info",
201        skip_all,
202        fields(application_id = %self.application_id),
203        err,
204    )]
205    /// # Errors
206    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
207    pub async fn run(self, builder: StreamsBuilder) -> Result<KafkaStreams, StreamsClientError> {
208        let built = builder.build(&self.application_id)?;
209        Box::pin(self.run_built(built)).await
210    }
211
212    /// Like [`run`](Self::run), but for a topology that the Processor API has
213    /// already built.
214    #[tracing::instrument(
215        name = "streams.app.run_built",
216        level = "info",
217        skip_all,
218        fields(application_id = %self.application_id, guarantee = ?self.processing_guarantee),
219        err,
220    )]
221    /// # Errors
222    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
223    pub async fn run_built(
224        self,
225        topology: BuiltTopology,
226    ) -> Result<KafkaStreams, StreamsClientError> {
227        self.cache
228            .prewarm()
229            .await
230            .map_err(|e| StreamsClientError::Runtime(e.to_string()))?;
231        KafkaStreams::builder()
232            .bootstrap(self.bootstrap)
233            .application_id(self.application_id)
234            .topology(topology)
235            .store_backend(self.store_backend)
236            .processing_guarantee(self.processing_guarantee)
237            .poll_interval(self.poll_interval.duration())
238            .commit_interval(self.commit_interval.duration())
239            .rebalance_timeout(self.rebalance_timeout.duration())
240            .join_retry_backoff(self.join_retry_backoff.duration())
241            .leave_heartbeat_timeout(self.leave_heartbeat_timeout.duration())
242            .broker_dns_timeout(self.broker_dns_timeout)
243            .client_dispatch_queue_capacity(self.client_dispatch_queue_capacity.get())
244            .client_frame_max(self.client_frame_max.size())
245            .fetch_min(self.fetch_min.size())
246            .cache_max_bytes(self.cache_max_bytes)
247            .interactive_query_queue_capacity(self.interactive_query_queue_capacity)
248            .maybe_barrier(self.barrier)
249            .build()
250            .await
251    }
252}
253
254#[cfg(test)]
255mod tests {
256    use assert2::check;
257
258    use super::*;
259
260    #[test]
261    fn cache_config_carries_schema_fetch_retry_policy() {
262        let policy = krabka_schema_serde::SchemaFetchRetryPolicy::new(
263            krabka_units::millis(37),
264            krabka_units::millis(91),
265        )
266        .unwrap();
267        let app = StreamsApp::builder()
268            .bootstrap("127.0.0.1:9092")
269            .application_id("schema-retry")
270            .schema_registry("http://127.0.0.1:8081")
271            .cache_config(krabka_schema_serde::CacheConfig {
272                fetch_retry_policy: policy,
273                ..Default::default()
274            })
275            .build();
276
277        check!(app.cache.fetch_retry_policy() == policy);
278    }
279
280    #[test]
281    fn state_store_cache_budget_preserves_raw_builder_default_and_override() {
282        let defaults = StreamsApp::builder()
283            .bootstrap("127.0.0.1:9092")
284            .application_id("cache-default")
285            .schema_registry("http://127.0.0.1:8081")
286            .build();
287        check!(defaults.cache_max_bytes == mebibytes(10));
288
289        let overridden = StreamsApp::builder()
290            .bootstrap("127.0.0.1:9092")
291            .application_id("cache-override")
292            .schema_registry("http://127.0.0.1:8081")
293            .cache_max_bytes(bytes(37))
294            .build();
295        check!(overridden.cache_max_bytes == bytes(37));
296    }
297
298    #[test]
299    fn interactive_query_queue_capacity_uses_typed_default_and_override() {
300        let defaults = StreamsApp::builder()
301            .bootstrap("127.0.0.1:9092")
302            .application_id("iq-capacity-default")
303            .schema_registry("http://127.0.0.1:8081")
304            .build();
305        assert_eq!(
306            defaults.interactive_query_queue_capacity,
307            crate::StreamsInteractiveQueryQueueCapacity::default()
308        );
309
310        let capacity =
311            crate::StreamsInteractiveQueryQueueCapacity::new(37).expect("positive queue capacity");
312        let overridden = StreamsApp::builder()
313            .bootstrap("127.0.0.1:9092")
314            .application_id("iq-capacity-override")
315            .schema_registry("http://127.0.0.1:8081")
316            .interactive_query_queue_capacity(capacity)
317            .build();
318        assert_eq!(overridden.interactive_query_queue_capacity, capacity);
319    }
320
321    #[test]
322    fn application_id_returns_configured_value() {
323        let app = StreamsApp::builder()
324            .bootstrap("127.0.0.1:9092")
325            .application_id("orders-app")
326            .schema_registry("http://127.0.0.1:8081")
327            .build();
328
329        assert_eq!(app.application_id(), "orders-app");
330    }
331
332    #[test]
333    fn broker_dns_timeout_uses_typed_default_and_override() {
334        let defaults = StreamsApp::builder()
335            .bootstrap("127.0.0.1:9092")
336            .application_id("default")
337            .schema_registry("http://127.0.0.1:8081")
338            .build();
339        assert_eq!(
340            defaults.broker_dns_timeout,
341            krabka_client_core::ClientDnsTimeout::default()
342        );
343
344        let timeout =
345            krabka_client_core::ClientDnsTimeout::new(millis(43)).expect("positive timeout");
346        let overridden = StreamsApp::builder()
347            .bootstrap("127.0.0.1:9092")
348            .application_id("override")
349            .schema_registry("http://127.0.0.1:8081")
350            .broker_dns_timeout(timeout)
351            .build();
352        assert_eq!(overridden.broker_dns_timeout, timeout);
353    }
354
355    #[test]
356    fn client_resource_policy_uses_typed_defaults_and_overrides() {
357        let dispatch =
358            krabka_client_core::ConnectionDispatchQueueCapacity::new(7).expect("positive capacity");
359        let frame = krabka_client_core::ClientFrameMax::try_from(kibibytes(32))
360            .expect("bounded frame maximum");
361        let fetch =
362            krabka_client_core::FetchMinBytes::try_from(bytes(3)).expect("positive fetch minimum");
363        let app = StreamsApp::builder()
364            .bootstrap("127.0.0.1:9092")
365            .application_id("client-policy")
366            .schema_registry("http://127.0.0.1:8081")
367            .client_dispatch_queue_capacity(dispatch)
368            .client_frame_max(frame)
369            .fetch_min(fetch)
370            .build();
371
372        assert_eq!(app.client_dispatch_queue_capacity, dispatch);
373        assert_eq!(app.client_frame_max, frame);
374        assert_eq!(app.fetch_min, fetch);
375    }
376
377    #[test]
378    fn runtime_cadence_uses_typed_defaults_and_independent_overrides() {
379        let defaults = StreamsApp::builder()
380            .bootstrap("127.0.0.1:9092")
381            .application_id("cadence-default")
382            .schema_registry("http://127.0.0.1:8081")
383            .build();
384        assert_eq!(
385            defaults.poll_interval,
386            crate::StreamsPollInterval::default()
387        );
388        assert_eq!(
389            defaults.commit_interval,
390            crate::StreamsCommitInterval::default()
391        );
392
393        let poll = crate::StreamsPollInterval::new(std::time::Duration::from_millis(37))
394            .expect("positive poll interval");
395        let commit = crate::StreamsCommitInterval::new(std::time::Duration::from_millis(41))
396            .expect("positive commit interval");
397        let overridden = StreamsApp::builder()
398            .bootstrap("127.0.0.1:9092")
399            .application_id("cadence-override")
400            .schema_registry("http://127.0.0.1:8081")
401            .poll_interval(poll)
402            .commit_interval(commit)
403            .build();
404        assert_eq!(overridden.poll_interval, poll);
405        assert_eq!(overridden.commit_interval, commit);
406    }
407
408    #[test]
409    #[allow(clippy::duration_suboptimal_units)]
410    fn rebalance_timeout_uses_typed_default_and_override() {
411        let defaults = StreamsApp::builder()
412            .bootstrap("127.0.0.1:9092")
413            .application_id("rebalance-default")
414            .schema_registry("http://127.0.0.1:8081")
415            .build();
416        assert_eq!(
417            defaults.rebalance_timeout,
418            crate::StreamsRebalanceTimeout::default()
419        );
420
421        let timeout = crate::StreamsRebalanceTimeout::new(std::time::Duration::from_millis(45_000))
422            .expect("valid timeout");
423        let overridden = StreamsApp::builder()
424            .bootstrap("127.0.0.1:9092")
425            .application_id("rebalance-override")
426            .schema_registry("http://127.0.0.1:8081")
427            .rebalance_timeout(timeout)
428            .build();
429        assert_eq!(overridden.rebalance_timeout, timeout);
430    }
431
432    #[test]
433    fn join_retry_backoff_uses_typed_default_and_override() {
434        let defaults = StreamsApp::builder()
435            .bootstrap("127.0.0.1:9092")
436            .application_id("join-retry-default")
437            .schema_registry("http://127.0.0.1:8081")
438            .build();
439        assert_eq!(
440            defaults.join_retry_backoff,
441            crate::StreamsJoinRetryBackoff::default()
442        );
443
444        let backoff = crate::StreamsJoinRetryBackoff::new(std::time::Duration::from_millis(37))
445            .expect("positive join retry backoff");
446        let overridden = StreamsApp::builder()
447            .bootstrap("127.0.0.1:9092")
448            .application_id("join-retry-override")
449            .schema_registry("http://127.0.0.1:8081")
450            .join_retry_backoff(backoff)
451            .build();
452        assert_eq!(overridden.join_retry_backoff, backoff);
453    }
454
455    #[test]
456    fn leave_heartbeat_timeout_uses_typed_default_and_override() {
457        let defaults = StreamsApp::builder()
458            .bootstrap("127.0.0.1:9092")
459            .application_id("leave-default")
460            .schema_registry("http://127.0.0.1:8081")
461            .build();
462        assert_eq!(
463            defaults.leave_heartbeat_timeout,
464            crate::StreamsLeaveHeartbeatTimeout::default()
465        );
466
467        let timeout =
468            crate::StreamsLeaveHeartbeatTimeout::new(std::time::Duration::from_millis(37))
469                .expect("positive timeout");
470        let overridden = StreamsApp::builder()
471            .bootstrap("127.0.0.1:9092")
472            .application_id("leave-override")
473            .schema_registry("http://127.0.0.1:8081")
474            .leave_heartbeat_timeout(timeout)
475            .build();
476        assert_eq!(overridden.leave_heartbeat_timeout, timeout);
477    }
478}