Struct KStream 

pub struct KStream<K, V, KS = <K as DefaultSerde>::Serde, VS = <V as DefaultSerde>::Serde> { /* private fields */ }

Implementations

impl<K, V, KS, VS> KStream<K, V, KS, VS>

pub fn with_key_serde<NewKS>(self, serde: NewKS) -> KStream<K, V, NewKS, VS>

pub fn with_value_serde<NewVS>(self, serde: NewVS) -> KStream<K, V, KS, NewVS>

impl<K, V, KS, VS> KStream<K, V, KS, VS>

where K: Any + Send + Sync + Clone, V: Any + Send + Clone, KS: Clone, VS: Clone,

pub fn map_values<V2, F>( &self, f: F, ) -> KStream<K, V2, KS, <V2 as DefaultSerde>::Serde>

where V2: DefaultSerde + Any + Send + Clone, F: Fn(&V) -> V2 + Clone + Send + Sync + 'static,

mapValues: transform each value, key unchanged. Not key-changing.

pub fn filter<F>(&self, predicate: F) -> KStream<K, V, KS, VS>

where K: Default, F: Fn(&K, &V) -> bool + Clone + Send + Sync + 'static,

filter: keep records where predicate(key, value) is true.

pub fn filter_not<F>(&self, predicate: F) -> KStream<K, V, KS, VS>

where K: Default, F: Fn(&K, &V) -> bool + Clone + Send + Sync + 'static,

filterNot: keep records where predicate(key, value) is false.

pub fn map<K2, V2, F>( &self, f: F, ) -> KStream<K2, V2, <K2 as DefaultSerde>::Serde, <V2 as DefaultSerde>::Serde>

where K: Default, K2: DefaultSerde + Any + Send + Clone, V2: DefaultSerde + Any + Send + Clone, F: Fn(&K, &V) -> (K2, V2) + Clone + Send + Sync + 'static,

map: transform key and value. Key-changing.

pub fn select_key<K2, F>( &self, f: F, ) -> KStream<K2, V, <K2 as DefaultSerde>::Serde, VS>

where K: Default, K2: DefaultSerde + Any + Send + Clone, F: Fn(&K, &V) -> K2 + Clone + Send + Sync + 'static,

selectKey: rewrite the key, value unchanged. Key-changing.

pub fn select_key_with_serde<K2, GKS, F>( &self, f: F, key_serde: GKS, ) -> KStream<K2, V, GKS, VS>

where K: Default, K2: Any + Send + Clone, GKS: Serde<K2> + Clone + 'static, F: Fn(&K, &V) -> K2 + Clone + Send + Sync + 'static,

selectKey with an explicit key serde. Key-changing.

pub fn flat_map<K2, V2, IT, F>( &self, f: F, ) -> KStream<K2, V2, <K2 as DefaultSerde>::Serde, <V2 as DefaultSerde>::Serde>

where K: Default, K2: DefaultSerde + Any + Send + Clone, V2: DefaultSerde + Any + Send + Clone, IT: IntoIterator<Item = (K2, V2)> + 'static, F: Fn(&K, &V) -> IT + Clone + Send + Sync + 'static,

flatMap: one record → zero or more (K2, V2). Key-changing.

pub fn flat_map_values<V2, IT, F>( &self, f: F, ) -> KStream<K, V2, KS, <V2 as DefaultSerde>::Serde>

where V2: DefaultSerde + Any + Send + Clone, IT: IntoIterator<Item = V2> + 'static, F: Fn(&V) -> IT + Clone + Send + Sync + 'static,

flatMapValues: one record → zero or more V2, key unchanged.

pub fn peek<F>(&self, f: F) -> KStream<K, V, KS, VS>

where K: Default, F: Fn(&K, &V) + Clone + Send + Sync + 'static,

peek: observe each record, then forward it unchanged.

pub fn foreach<F>(self, f: F)

where K: Default, F: Fn(&K, &V) + Clone + Send + Sync + 'static,

foreach: terminal side-effect on each record (consumes the stream).

pub fn to_explicit<KS2, VS2>( self, topic: impl Into<String>, produced: impl Into<Produced<KS2, VS2>>, )

where KS2: SerdeAssociate + Serde<K> + Clone, VS2: SerdeAssociate + Serde<V> + Clone,

to: write the stream to a topic through a sink (consumes the stream).

pub fn to(self, topic: impl Into<String>)

where KS: SerdeAssociate + Serde<K> + Clone, VS: SerdeAssociate + Serde<V> + Clone,

Write the stream to a topic using the default/carried serdes of the stream.

pub fn merge(&self, other: &KStream<K, V, KS, VS>) -> KStream<K, V, KS, VS>

merge: union this stream with other (same K/V). Both feed one node.

pub fn join_table<VT, VO, F, VTS>( &self, table: &KTable<K, VT, KS, VTS>, joiner: F, ) -> KStream<K, VO, KS, <VO as DefaultSerde>::Serde>

where VT: Any + Send + Clone, VO: DefaultSerde + Any + Send + Clone, VTS: Clone, F: Fn(&V, &VT) -> VO + Clone + Send + Sync + 'static,

join (inner stream-table join): look each stream record up in table.

For each record on this stream, this method looks the record’s key up in table’s materialized store. Only when a value is present, it forwards joiner(&stream_value, &table_value) keyed by the same key. It drops records whose key is absent from the table. This is an inner join.

table must be materialized, that is, sourced through StreamsBuilder::table or any materialized op. The join reads its store by name. The join declares the stream side and the table’s source topic as a copartition group (KIP-1071), so the streams-group coordinator co-locates their partitions.

The stream key must be unchanged relative to its source partitioning, because a stream-table join is partition-local. If a key-changing op such as map, select_key, flat_map, or group_by comes before the join, call repartition first to re-partition by the new key. If you do not, join_table panics.

Naming vs the JVM

The JVM overloads join for both stream-table and (windowed) stream-stream joins. Rust cannot have two inherent methods of the same name and differing arity, so the stream-table forms are named join_table/left_join_table here, leaving the plain join/left_join for the windowed stream-stream join (see KStream::join).

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn left_join_table<VT, VO, F, VTS>( &self, table: &KTable<K, VT, KS, VTS>, joiner: F, ) -> KStream<K, VO, KS, <VO as DefaultSerde>::Serde>

where VT: Any + Send + Clone, VO: DefaultSerde + Any + Send + Clone, VTS: Clone, F: Fn(&V, Option<&VT>) -> VO + Clone + Send + Sync + 'static,

leftJoin (left stream-table join): like join_table but always forwards a record for every stream record. On a table miss the joiner receives None for the table-side value. See join_table for the naming-vs-JVM note.

pub fn join_table_with<VT, VO, F, VTS>( &self, table: &KTable<K, VT, KS, VTS>, joiner: F, join_config: Joined, ) -> KStream<K, VO, KS, <VO as DefaultSerde>::Serde>

where VT: Any + Send + Sync + Clone, VO: DefaultSerde + Any + Send + Clone, VTS: Clone, KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static, V: Sync, F: Fn(&V, &VT) -> VO + Clone + Send + Sync + 'static,

join with Joined config (KIP-923 grace path).

This method is identical to join_table when joined.grace is None. When a grace period is set, the join buffers each stream record into a JoinGraceBufferStore. It drains the buffer as-of the record’s own timestamp once the grace horizon passes. Out-of-order stream records therefore join against the table version that was current at the record’s own timestamp.

Grace requires the table to be versioned. See Materialized::as_versioned. grace must be strictly less than the table’s history_retention.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn left_join_table_with<VT, VO, F, VTS>( &self, table: &KTable<K, VT, KS, VTS>, joiner: F, join_config: Joined, ) -> KStream<K, VO, KS, <VO as DefaultSerde>::Serde>

where VT: Any + Send + Sync + Clone, VO: DefaultSerde + Any + Send + Clone, VTS: Clone, KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static, V: Sync, F: Fn(&V, Option<&VT>) -> VO + Clone + Send + Sync + 'static,

leftJoin with Joined config (KIP-923 grace path). Like left_join_table but, when joined.grace is set, wires the grace buffer + as-of drain (see join_table_with). On a table miss at drain time the joiner receives None.

pub fn join_global<GK, VG, VR, KM, J, GKS, VGS>( &self, global: &GlobalKTable<GK, VG, GKS, VGS>, key_mapper: KM, joiner: J, ) -> KStream<K, VR, KS, <VR as DefaultSerde>::Serde>

where GK: Any + Send + Sync + 'static, VG: Any + Send + Clone, VR: DefaultSerde + Any + Send + Clone, GKS: Clone, VGS: Clone, KM: Fn(&K, &V) -> GK + Clone + Send + Sync + 'static, J: Fn(&V, &VG) -> VR + Clone + Send + Sync + 'static,

join (inner stream-globaltable join): look each record up in global.

For each record on this stream, this method computes the lookup key gk = key_mapper(&streamKey, &streamValue) and looks it up in global’s fully-replicated store. Only on a hit, it forwards joiner(&stream_value, &global_value) keyed by the stream key with the stream timestamp. It drops records whose derived key is absent.

The lookup key is derived from the record, unlike join_table, so it can differ from the stream key. A GlobalKTable is fully replicated, so there is no copartitioning, no repartition, and no key-changing assertion. Any record can look up any key on every instance.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn left_join_global<GK, VG, VR, KM, J, GKS, VGS>( &self, global: &GlobalKTable<GK, VG, GKS, VGS>, key_mapper: KM, joiner: J, ) -> KStream<K, VR, KS, <VR as DefaultSerde>::Serde>

where GK: Any + Send + Sync + 'static, VG: Any + Send + Clone, VR: DefaultSerde + Any + Send + Clone, GKS: Clone, VGS: Clone, KM: Fn(&K, &V) -> GK + Clone + Send + Sync + 'static, J: Fn(&V, Option<&VG>) -> VR + Clone + Send + Sync + 'static,

leftJoin (left stream-globaltable join): like join_global but always forwards a record for every stream record. On a global-store miss the joiner receives None for the global-side value.

pub fn join<V2, VO, F, V2S>( &self, other: &KStream<K, V2, KS, V2S>, joiner: F, windows: JoinWindows, stream_joined: StreamJoined<KS, VS, V2S>, ) -> KStream<K, VO, KS, <VO as DefaultSerde>::Serde>

where V2: Any + Send + Sync + Clone, VO: DefaultSerde + Any + Send + Clone, F: Fn(&V, &V2) -> VO + Clone + Send + Sync + 'static, KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static, V2S: Serde<V2> + Clone + 'static, V: Send + Sync,

join (windowed inner stream-stream join): match records over a window.

For each record on either stream, the join puts the record into its own JoinWindowStore and fetches the OTHER stream’s store over the join window. It forwards one output per match, keyed by the same key. A left record at t matches right records with a timestamp in [t - before, t + after]. The rule is symmetric: the per-side OTHER processor swaps before and after, so the rule holds whichever side drives the record.

The join lowers to a THIS processor fed by this stream, an OTHER processor fed by other, and a MERGE node that unions the two outputs. This mirrors KTable::join’s dual+merge shape. Each side puts into its own retainDuplicates join-window store and reads the other side’s store. Each side connects to BOTH stores, so the grouping pass folds A, B, both stores, and the two join nodes into one copartitioned subtopology. When both streams are single-source-topic streams, the join declares their source topics as a copartition group (KIP-1071).

Both stream keys must be unchanged relative to their source partitioning. You must call repartition() after a key-changing op upstream. If you do not, the join panics.

Naming vs the JVM

The JVM overloads join for stream-table and stream-stream. Rust cannot, so the windowed stream-stream join keeps join/left_join while the stream-table forms are join_table/ left_join_table.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn left_join<V2, VO, F, V2S>( &self, other: &KStream<K, V2, KS, V2S>, joiner: F, windows: JoinWindows, stream_joined: StreamJoined<KS, VS, V2S>, ) -> KStream<K, VO, KS, <VO as DefaultSerde>::Serde>

where V2: Any + Send + Sync + Clone, VO: DefaultSerde + Any + Send + Clone, F: Fn(&V, Option<&V2>) -> VO + Clone + Send + Sync + 'static, KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static, V2S: Serde<V2> + Clone + 'static, V: Send + Sync,

leftJoin (windowed left stream-stream join): emit unmatched left records.

Every record on THIS (left) stream that finds no match in the OTHER (right) window emits joiner(&this, None) once that record’s window closes. Stream time drives the window close (KIP-633). Matched records emit joiner(&this, Some(&other)), as in the inner join. The right side never emits a non-join.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn outer_join<V2, VO, F, V2S>( &self, other: &KStream<K, V2, KS, V2S>, joiner: F, windows: JoinWindows, stream_joined: StreamJoined<KS, VS, V2S>, ) -> KStream<K, VO, KS, <VO as DefaultSerde>::Serde>

where V2: Any + Send + Sync + Clone, VO: DefaultSerde + Any + Send + Clone, F: Fn(Option<&V>, Option<&V2>) -> VO + Clone + Send + Sync + 'static, KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static, V2S: Serde<V2> + Clone + 'static, V: Send + Sync,

outerJoin (windowed outer stream-stream join): emit from either side.

Every record on EITHER side that finds no match emits joiner(Some, None) or joiner(None, Some) once its window closes (KIP-633). Matched records emit joiner(Some, Some).

pub fn repartition<KS2, VS2>( &self, repartitioned: Repartitioned<KS2, VS2>, ) -> KStream<K, V, KS2, VS2>

where KS2: Serde<K> + Clone + 'static, VS2: Serde<V> + Clone + 'static,

repartition: force a repartition through an internal topic.

This op lowers as sink → add_repartition_topic → source. It is the same pattern that the implicit repartition before a stateful aggregation uses. The repartition topic name is <app_id>-<name>-repartition. <name> is the explicit Repartitioned name when set, otherwise an auto-name minted from the counter.

Byte-exactness vs JVM: the JVM assigns a distinct KSTREAM-REPARTITION- counter for standalone repartition() calls. No golden fixture validates that counter. The bar here is functional correctness: no panic, and records flow through.

pub fn split(&self) -> BranchedStream<K, V, KS, VS>

split: begin a branching fan-out.

Returns a BranchedStream builder. Individual branch calls on that builder create filtered child streams. The split itself adds no node to the topology. Each branch call creates a filter-backed child wired directly to this stream’s node.

Simplification vs JVM: each branch receives a record when its predicate matches, not just the first matching branch. For mutually-exclusive predicates the behaviour is identical to the JVM first-match-wins semantics.

pub fn group_by_key_explicit<GKS, GVS>( &self, grouped: impl Into<Grouped<GKS, GVS>>, ) -> KGroupedStream<K, V>

where GKS: Serde<K> + Clone + 'static, GVS: Serde<V> + Clone + 'static,

groupByKey: group by the existing key, ready for an aggregation.

This op records no graph node. A terminal count, reduce, or aggregate call records the optional repartition and the aggregate node. The returned KGroupedStream carries two things: whether the upstream key lineage is key-changing, which forces the aggregation to insert a repartition, and a typed repartition-lowering thunk built from the Grouped serdes.

pub fn group_by_key(&self) -> KGroupedStream<K, V>

where KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static,

groupByKey using the stream’s existing serdes.

pub fn group_by<GKS, GVS, F>( &self, f: F, grouped: impl Into<Grouped<GKS, GVS>>, ) -> KGroupedStream<GKS::Target, V>

where K: Default, GKS: SerdeAssociate + Serde<GKS::Target> + Clone + 'static, GVS: SerdeAssociate + Serde<V> + Clone + 'static, GKS::Target: Any + Send + Sync + Clone, F: Fn(&K, &V) -> GKS::Target + Clone + Send + Sync + 'static,

groupBy: re-key with f, then group by the new key.

This op is equivalent to select_key(f).group_by_key_explicit(grouped). The key change forces a repartition before any later aggregation.

pub fn process<KOut, VOut, PS>( &self, supplier: PS, store_names: impl IntoIterator<Item = impl Into<String>>, ) -> KStream<KOut, VOut, <KOut as DefaultSerde>::Serde, <VOut as DefaultSerde>::Serde>

where KOut: DefaultSerde + Any + Send + Sync + Clone, VOut: DefaultSerde + Any + Send + Clone, PS: ProcessorSupplier<K, V, KOut, VOut> + Clone + 'static,

process: attach a custom Processor-API node that may rewrite the key.

supplier is any user ProcessorSupplier, for example a || MyProc closure. store_names names the stores that StreamsBuilder::add_state_store registered and that the processor reads and writes through ProcessorContext::get_state_store. This method looks up each named store’s connect thunk now, so you must add the store before this call. Lowering then invokes each thunk to register the store, emit its compact <app>-<store>-changelog changelog, and connect the store to this processor node.

This op mirrors the JVM KStream.process. It treats the result as key-changing, because the processor may call forward with any key. The single-source-topic lineage therefore breaks, and a downstream aggregation or join must repartition first.

Panics

Panics if any name in store_names was not registered through add_state_store.

pub fn process_values<VOut, PS>( &self, supplier: PS, store_names: impl IntoIterator<Item = impl Into<String>>, ) -> KStream<K, VOut, KS, <VOut as DefaultSerde>::Serde>

where VOut: DefaultSerde + Any + Send + Clone, PS: FixedKeyProcessorSupplier<K, V, VOut> + Clone + 'static,

processValues: attach a custom fixed-key Processor-API node (KIP-820).

The node may rewrite the value but NOT the key.

supplier is any FixedKeyProcessorSupplier, for example a || MyFixedProc closure. store_names names the stores that add_state_store registered and that the processor reads and writes through FixedKeyProcessorContext::get_state_store. As with process, this method looks up each named store’s connect thunk now, so you must add the store before this call. Lowering then invokes each thunk to register the store, emit its compact <app>-<store>-changelog changelog, and connect the store to this processor node.

Unlike process, this op keeps the key. The result is therefore non-key-changing and carries the single-source-topic lineage unchanged, so a downstream aggregation or join needs no repartition. An internal adapter wraps the supplier, so the runtime sees a Processor<K, V, K, VOut>.

Panics

Panics if any name in store_names was not registered through add_state_store.

pub fn to_table_explicit<NKS, NVS>( &self, materialized: impl Into<Materialized<NKS, NVS>>, ) -> KTable<K, V, NKS, NVS>

where NKS: Serde<K> + Clone + 'static, NVS: Serde<V> + Clone + 'static,

toTable: materialize this stream into a KTable.

The op writes each record into a state store and forwards a Change<V> change-stream that carries the prior store value as old.

to_table carries the key through unchanged, so it never inserts a repartition. The JVM repartitions only when the upstream key is rewritten without a re-group. The store name is Materialized’s explicit name when set, otherwise a fresh KSTREAM-TOTABLE-STATE-STORE- counter. The store gets the standard <app>-<store>-changelog changelog, or no changelog when Materialized::with_logging(false).

pub fn to_table(&self, store_name: impl Into<String>) -> KTable<K, V, KS, VS>

where KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static,

Sourced KTable from this stream using the stream’s carried serdes.