Struct Topology 

pub struct Topology { /* private fields */ }

A Processor-API topology under construction.

Node insertion order is significant. It sets the subtopology indices, which match the JVM.

Implementations

impl Topology

pub fn new() -> Self

pub fn add_source<K, V>( &mut self, name: impl Into<String>, topics: impl IntoIterator<Item = impl Into<String>>, ) -> NodeHandle<K, V>

where K: Any + Send + Sync + Clone + DefaultSerde, V: Any + Send + Sync + Clone + DefaultSerde,

Add a source node reading the given external topics with the default serdes for K and V, returning a typed NodeHandle used to wire children to it.

Use Topology::add_source_explicit when either type has no default serde or when a topology needs custom serdes.

pub fn add_source_explicit<K, V, KS, VS>( &mut self, name: impl Into<String>, topics: impl IntoIterator<Item = impl Into<String>>, consumed: impl Into<Consumed<KS, VS>>, ) -> NodeHandle<K, V>

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

Add a source node reading the given external topics with explicit serdes, returning a typed NodeHandle used to wire children to it.

consumed carries the key serde and the value serde. At runtime they deserialize incoming bytes into typed Record<K, V> values. Write it as Consumed::with(key_serde, value_serde) so the two roles are visible.

Prefer Topology::add_source, the default-serde form, for types that implement DefaultSerde. Use this escape hatch when a type has no default serde, or to override the default for a topic. Examples are a hand-rolled Serde<T>, a key-role schema serde (AvroSerde::<T>::key(&cache)), and a validation-on JSON serde (JsonSerde::value(&cache, true)).

pub fn add_processor<KIn, VIn, KOut, VOut, S, P, I>( &mut self, name: impl Into<String>, supplier: S, parents: I, ) -> NodeHandle<KOut, VOut>

where KIn: Any + Send, VIn: Any + Send, KOut: Any + Send + Clone, VOut: Any + Send + Clone, S: ProcessorSupplier<KIn, VIn, KOut, VOut> + Clone, I: IntoIterator<Item = P>, P: Borrow<NodeHandle<KIn, VIn>>,

Add a processor node fed by the given parent NodeHandles, returning a handle for its output.

Wiring is by value. Every parent’s output type must equal this processor’s input type (KIn, VIn), and the compiler enforces this. supplier produces a fresh Processor per task. The closure form || MyProc satisfies ProcessorSupplier through a blanket impl, and it infers all four KV type parameters from the processor’s Processor impl, so callers never annotate them.

pub fn add_sink<K, V>( &mut self, name: impl Into<String>, topic: impl Into<String>, parents: impl IntoIterator<Item = impl Borrow<NodeHandle<K, V>>>, )

where K: Any + Send + Sync + DefaultSerde, V: Any + Send + Sync + DefaultSerde,

Add a sink node writing to topic with the default serdes for K and V, fed by the given parent NodeHandles. Every parent’s output type must equal the sink’s input type (K, V), and the compiler enforces this.

Use Topology::add_sink_explicit when either type has no default serde or when a topology needs custom serdes.

pub fn add_sink_explicit<K, V, KS, VS, P, I>( &mut self, name: impl Into<String>, topic: impl Into<String>, parents: I, produced: impl Into<Produced<KS, VS>>, )

where K: Any + Send, V: Any + Send, KS: Serde<K> + Clone, VS: Serde<V> + Clone, I: IntoIterator<Item = P>, P: Borrow<NodeHandle<K, V>>,

Add a sink node writing to topic with explicit serdes, fed by the given parent NodeHandles. Every parent’s output type must equal the sink’s input type (K, V), and the compiler enforces this.

produced carries the key serde and the value serde that serialize outgoing records. Write it as Produced::with(key_serde, value_serde). A sink is terminal, so this method returns nothing.

Prefer Topology::add_sink, the default-serde form, for types that implement DefaultSerde. Use this escape hatch when a type has no default serde, or to override the default for a topic. Examples are a hand-rolled Serde<T>, a key-role schema serde (AvroSerde::<T>::key(&cache)), and a validation-on JSON serde (JsonSerde::value(&cache, true)).

pub fn add_state_store<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, processors: impl IntoIterator<Item = impl Into<String>>, ) -> &mut Self

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

Register a state store connected to the given processors, with a changelog.

key_serde and value_serde define how the store serializes records into the changelog topic (<app_id>-<name>-changelog) and into the store’s byte map. Stores connect processors that can have different types, so the processor list uses names. Pass NodeHandle::name for a handle you hold.

pub fn add_state_store_with_changelog<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, processors: impl IntoIterator<Item = impl Into<String>>, changelog_topic: impl Into<String>, ) -> &mut Self

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

Register a state store whose changelog is an existing source topic rather than the derived <app_id>-<name>-changelog.

This backs the REUSE_KTABLE_SOURCE_TOPICS DSL optimizer. A builder.table_explicit(topic, ...) store can reuse topic as its changelog. No separate app-<store>-changelog topic is created, and the wire topology lists topic as the store’s changelog. changelog_topic is the topic name used both in the wire state_changelog_topics entry and as the runtime store’s changelog target.

pub fn add_window_store<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, window: (Time, Time, Time), processors: impl IntoIterator<Item = impl Into<String>>, ) -> &mut Self

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

Register a windowed state store connected to the given processors.

Like add_state_store but for windowed stores. The changelog topic carries compact,delete configs and a retention.ms derived from size + grace + [CHANGELOG_ADDITIONAL_RETENTION].

The tuple is (retention_basis, window_size, grace). The retention basis sets the changelog retention.ms. For tumbling and hopping windows it equals the window size. For sliding windows it is the retention span (2 * time_difference), which is wider than the window.

The window size is the true window length. Only the store’s cache flush uses it, to rebuild the downstream Windowed<K> key’s end = start + window_size, because the store-key bytes hold only the start. For sliding windows this is time_difference (1x), NOT the retention span.

pub fn add_join_window_store<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, window: (Time, Time, Time), processors: impl IntoIterator<Item = impl Into<String>>, ) -> &mut Self

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

Register a join window state store connected to the given processors.

Like add_window_store but for join window stores (retainDuplicates). The changelog topic carries delete-only configs and a retention.ms derived from before + after + grace + [CHANGELOG_ADDITIONAL_RETENTION]. Compaction is not applicable because the store retains duplicates.

pub fn add_session_store<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, gap: Time, grace: Time, processors: impl IntoIterator<Item = impl Into<String>>, ) -> &mut Self

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

Register a session state store connected to the given processors.

Like add_window_store but for session stores. It reuses the windowed (compact,delete) changelog config. The retention.ms comes from gap + grace + [CHANGELOG_ADDITIONAL_RETENTION]. The store holds the raw aggregate (SessionBytesStore).

pub fn add_versioned_store<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, history_retention: Time, processors: impl IntoIterator<Item = impl Into<String>>, ) -> &mut Self

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

Register a versioned state store (KIP-889) connected to the given processors.

The changelog topic carries compact + min.compaction.lag.ms = history_retention + [CHANGELOG_ADDITIONAL_RETENTION]. The version-chain store is self-contained in memory, so it does not use the supplied byte backend.

pub fn add_versioned_store_with_changelog<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, history_retention: Time, processors: impl IntoIterator<Item = impl Into<String>>, changelog_topic: impl Into<String>, ) -> &mut Self

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

Like add_versioned_store but the changelog is an existing source topic. The REUSE_KTABLE_SOURCE_TOPICS optimizer points a versioned builder.table_explicit(topic, …) store’s changelog at its own source topic. This mirrors add_state_store_with_changelog.

pub fn add_suppress_store<K, V, KS, VS>( &mut self, name: impl Into<String>, key_serde: KS, value_serde: VS, logging: bool, processors: impl IntoIterator<Item = impl Into<String>>, ) -> &mut Self

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

Register a suppress buffer store connected to the given processor.

The suppress buffer (SuppressBytesStore) is a time-ordered in-memory buffer with its own storage. It does NOT use the pluggable byte backend, so the factory ignores the opened backend.

logging toggles ONLY the changelog. When true, the wire topology gets the changelog topic, a plain cleanup.policy=compact changelog, because the JVM suppress buffer is a compacted KV store, and the store logs and restores. When false, the store stays in memory and NO changelog topic appears, so a logging-off suppress has the same wire shape as an unsuppressed topology.

pub fn add_global_store<K, V, KS, VS>( &mut self, store_name: impl Into<String>, source_name: impl Into<String>, topic: impl Into<String>, processor_name: impl Into<String>, consumed: impl Into<Consumed<KS, VS>>, ) -> &mut Self

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

Register a GlobalKTable source, update-processor, and KV store.

A GlobalKTable is invisible in the wire. It has no subtopology of its own and no changelog topic. Its source node still takes a node-group index during grouping, so other subtopology ids shift. This method does three things:

1. it registers a source node reading topic and a processor node fed by it, and the source→processor edge unites them into one node group;
2. it marks topic global so the grouping pass skips it in the source-bucketing pass. The final filter drops the resulting source-less group, but that group already took its index;
3. it stores the global KV factory in a SEPARATE map (global_store_factories, NOT store_factories) so per-task instantiate does not build it and NO changelog topic is emitted. The factory builds a KeyValueBytesStore with an empty changelog.

Per-task instantiate does not build the global store. The fully-replicated global-store manager reads the factory.

pub fn connect_processor_store( &mut self, processor: &str, store: &str, ) -> &mut Self

Connect an additional processor to an already-registered state store.

This mirrors InternalTopologyBuilder.connectProcessorAndStateStores. A join processor must read the joined table’s store even though it was not the store’s original owner. The grouping pass unions all connected processors into one subtopology, so a second processor here is enough to pull the join processor into the same subtopology as the store.

pub fn add_repartition_topic<S: Into<String>>(&mut self, name: S) -> &mut Self

Register a topic name as an internal repartition topic.

pub fn add_copartition_group( &mut self, topics: impl IntoIterator<Item = impl Into<String>>, ) -> &mut Self

Declare a copartition group. The given member topics must share a partitioning.

The grouping pass assigns the group to the subtopology that reads its members. The wire layer encodes member names as int16 indices into that subtopology’s sorted source and repartition arrays. Joins need this (KIP-1071).

pub fn build<S: Into<String>>( self, application_id: S, ) -> Result<BuiltTopology, TopologyError>

Derive subtopologies and the wire topology.

application_id sets the internal-topic names (<app>-<store>-changelog).

The typed NodeHandle wiring already guarantees that parent→child KV types match, so build() checks only the structural invariants: no duplicate names, every predecessor exists, and at least one source. The wire Topology is byte-identical to the untyped implementation.

Errors

Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.

Trait Implementations

impl Debug for Topology

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for Topology

fn default() -> Topology

Returns the “default value” for a type.