Struct TopologyTestDriver 

pub struct TopologyTestDriver { /* private fields */ }

Synchronous test driver.

This driver builds a runnable processor graph from a BuiltTopology. It exposes pipe_input and read_output, so a test can exercise processing logic without a real broker.

Implementations

impl TopologyTestDriver

pub fn new(built: &BuiltTopology) -> Result<Self, ProcessorError>

Instantiates the topology’s graph for testing.

Errors

Returns the ProcessorError that instantiation or a processor’s init raises. A topology that build accepted raises none in practice.

pub fn pipe_input<KS, VS>( &mut self, topic: &str, consumed: impl Into<Consumed<KS, VS>>, key: Option<KS::Target>, value: VS::Target, timestamp: i64, )

where KS: SerdeAssociate + Serde<KS::Target>, VS: SerdeAssociate + Serde<VS::Target>,

Serializes and pipes one record on topic, then loops repartition outputs back.

consumed is the same Consumed::with(key_serde, value_serde) pair the source for topic reads with.

pub fn advance_wall_clock_time(&mut self, by: Duration)

Advances the mock wall clock by by and fires wall-clock punctuators.

Each punctuator fires at most once, with the new clock as its value. This mirrors the JVM TopologyTestDriver.advanceWallClockTime.

pub fn get_key_value_store<K: Send + Sync + 'static, V: Send + 'static>( &mut self, name: &str, ) -> Option<&mut dyn KeyValueStore<K, V>>

Inspects a state store’s contents after piping.

This method mirrors the JVM TopologyTestDriver.getKeyValueStore. It is test-only. Do not use it for interactive queries.

pub fn store_get<K: Send + Sync + 'static, V: Send + 'static>( &mut self, store: &str, key: &K, ) -> Option<V>

Test-only synchronous store read that calls block_on on the async store.

The store API is async. This method drives one typed get to completion, so sync #[test] assertions can inspect materialized state.

pub fn store_get_versioned<K: Send + Sync + 'static, V: Send + 'static>( &mut self, store_name: &str, key: &K, ) -> Option<V>

Latest live value of a key in a versioned store. This is a test helper.

pub fn drain_changelog( &mut self, ) -> Vec<(String, Bytes, Option<Bytes>, Option<i64>)>

Drains the changelog records the stores have produced so far. This is a test helper.

The records come back as (changelog_topic, key, value, record_ts) in production order. A test can then assert byte-exact changelog records, for example the KIP-889 versioned changelog, which holds bare value bytes and the version timestamp in the record-timestamp field.

pub fn pipe_global<K, V>(&mut self, store_name: &str, key: K, value: V)

where K: Send + Sync + 'static, V: Send + 'static,

Populates a GLOBAL store directly. This method is test-only.

In a real app the global consumer reads all partitions of the source topic and applies them to the fully-replicated global store. The test driver instead injects values straight into the shared global store manager, so a stream-globaltable join can look them up.

pub async fn iq_kv_get<K: 'static, V: 'static>( &self, store: &str, key: &K, ks: &dyn Serde<K>, vs: &dyn Serde<V>, ) -> Option<V>

Interactive-query KV read: the value for key, else None. The result is None if the store is absent or is not a key-value store.

Panics

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

pub async fn iq_kv_range<K: 'static, V: 'static>( &self, store: &str, lo: &K, hi: &K, ks: &dyn Serde<K>, vs: &dyn Serde<V>, ) -> Vec<(K, V)>

Interactive-query KV range read over the inclusive [lo, hi] key span, in store (memcmp) order. The result is empty if the store is absent or is not a KV store.

Panics

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

pub async fn iq_kv_all<K: 'static, V: 'static>( &self, store: &str, ks: &dyn Serde<K>, vs: &dyn Serde<V>, ) -> Vec<(K, V)>

Interactive-query read of every entry in a KV store, in store order.

Panics

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

pub async fn iq_kv_count(&self, store: &str) -> u64

Interactive-query approximate entry count for a KV store. The result is 0 if the store is absent.

pub async fn iq_window_fetch<K: 'static, V: 'static>( &self, store: &str, key: &K, from: i64, to: i64, ks: &dyn Serde<K>, vs: &dyn Serde<V>, ) -> Vec<(i64, V)>

Interactive-query window read: (windowStart, value) for every window of key whose start is in the inclusive [from, to] span, ascending by window start. The result is empty if the store is absent or is not a window store.

Panics

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

pub async fn iq_window_fetch_single<K: 'static, V: 'static>( &self, store: &str, key: &K, window_start: i64, ks: &dyn Serde<K>, vs: &dyn Serde<V>, ) -> Option<V>

Interactive-query single-window read: value of the window of key that starts exactly at window_start, else None.

Panics

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

pub async fn query<Q: Query>( &self, req: StateQuery<Q>, ) -> StateQueryResult<Q::Result>

Runs an IQv2 query against the single test graph, partition 0.

Position is empty, because the driver does not track source offsets. Use unit tests on Position::dominates for bound behavior.

pub async fn iq_session_fetch<K: 'static, V: 'static>( &self, store: &str, key: &K, ks: &dyn Serde<K>, vs: &dyn Serde<V>, ) -> Vec<((i64, i64), V)>

Interactive-query session read: ((start, end), value) for every session of key, in store order. The result is empty if the store is absent or is not a session store.

Panics

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

pub fn read_output<KS, VS>( &mut self, topic: &str, produced: impl Into<Produced<KS, VS>>, ) -> Option<(Option<KS::Target>, VS::Target)>

where KS: SerdeAssociate + Serde<KS::Target>, VS: SerdeAssociate + Serde<VS::Target>,

Pops and deserializes the next output record for topic.

produced is the same Produced::with(key_serde, value_serde) pair the sink writing topic produced with.

Panics

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