Struct EmbeddedTask 

pub struct EmbeddedTask { /* private fields */ }

One task’s processor graph, driven by an embedder instead of the runtime.

The graph runs with in-memory stores and with the record cache disabled, which is the JVM TopologyTestDriver default, so a store emits on every update. Changelog topics derive from the built topology’s application id, exactly as BuiltTopology::changelog_topics lists them.

A KIP-1071 task is one (subtopology, partition) of the group’s assignment, and for_subtopology instantiates exactly that subtopology’s graph. new instantiates every subtopology, which is the task itself when the topology has one.

Examples

use assert2::assert;
use krabka_client_streams::{EmbeddedTask, NodeHandle, Record, Topology, impl_processor};

struct Upper;
impl_processor! {
    impl Upper: (String, String) -> (String, String) {
        async fn process(&mut self, ctx, r) {
            ctx.forward(Record::new(r.key, r.value.to_uppercase(), r.timestamp));
        }
    }
}

let mut topo = Topology::new();
let src: NodeHandle<String, String> = topo.add_source("src", ["in"]);
let up = topo.add_processor("up", || Upper, [&src]);
topo.add_sink("out", "out", [&up]);
let built = topo.build("app").unwrap();

let mut task = EmbeddedTask::new(&built).unwrap();
task.pipe("in", 0, 0, Some(b"k"), b"hello", 0).unwrap();
let out = task.take_output();
assert!(out[0].value.as_deref() == Some(&b"HELLO"[..]));

Implementations

impl EmbeddedTask

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

Instantiates the whole topology’s graph, every subtopology included, with in-memory stores, and calls init on every processor.

Use it for a topology with one subtopology, where the task is the topology. A KIP-1071 task of a topology with several subtopologies must use for_subtopology: a graph that holds every subtopology runs every branch that reads a piped topic, so two tasks of different subtopologies over the same source topic would each emit the other’s sink output and store updates, and each would hold the other’s stores and fire its punctuators.

Errors

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

pub fn for_subtopology( built: &BuiltTopology, subtopology_id: &str, ) -> Result<Self, ProcessorError>

Instantiates the graph of one subtopology, with in-memory stores, and calls init on its processors.

The graph holds that subtopology’s sources, processors and sinks, and the stores its processors connect to, and nothing of any other subtopology. This is the graph of a KIP-1071 task (subtopology_id, partition); BuiltTopology::subtopology_ids lists the ids and BuiltTopology::source_topics_for the topics to pipe.

Errors

Returns ProcessorError::UnknownSubtopology when the topology has no subtopology subtopology_id, else what new returns.

pub fn pipe( &mut self, topic: &str, partition: i32, offset: i64, key: Option<&[u8]>, value: &[u8], timestamp: i64, ) -> Result<(), ProcessorError>

Runs one source record through the graph.

partition and offset are the record’s provenance, and a processor reads them through its record context. The record raises the task’s stream-time to its timestamp when it is later. A record on a topic no source reads is ignored. The sink outputs wait in take_output and the store changelog records in drain_changelogs.

Errors

Returns a ProcessorError when a source serde cannot deserialize the record, or a sink serde cannot serialize an output.

pub fn take_output(&mut self) -> Vec<OutputRecord>

Takes the sink records the graph emitted since the last call, in emission order.

A record for a repartition topic is in the list too. The embedder produces it like any other output, and the subtopology that reads the repartition topic gets it back through pipe.

pub fn drain_changelogs(&mut self) -> Vec<ChangelogRecord>

Takes the changelog records the stores logged since the last call.

A reuse-source store’s records are dropped, not returned, because its changelog topic is the source topic itself. Restore writes log nothing while logging is off.

pub fn restore_apply( &mut self, store: &str, key: Bytes, value: Option<Bytes>, timestamp: i64, )

Applies one changelog record to the named store, as a restore does.

A value of None is a tombstone. timestamp is the changelog record’s timestamp; a versioned store inserts the version at it. Turn logging off with set_logging for the whole replay, so the restored writes are not logged again, and turn it back on before the first pipe. A store the graph does not hold is ignored.

pub fn apply_global(&mut self, store: &str, key: Bytes, value: Option<Bytes>)

Applies one record of a global source topic to the named global store.

A GlobalKTable store is fully replicated and has no changelog, so the embedder replays every partition of the source topic here, before the first record that joins against it. A value of None deletes the entry. BuiltTopology::global_store_topics maps each store to its topic.

pub fn set_logging(&mut self, on: bool)

Turns changelog logging on or off for every store.

pub fn stream_time(&self) -> i64

The task’s stream-time: the highest record timestamp piped so far, or i64::MIN before the first record.

pub fn punctuate_stream_time( &mut self, stream_time: i64, ) -> Result<(), ProcessorError>

Raises stream-time to stream_time when it is higher, then fires every due stream-time punctuator once at the current stream-time.

The runtime fires stream-time punctuators after each fetched batch. Passing stream_time fires what is due without moving the clock. What a punctuator forwards waits in take_output and drain_changelogs.

Errors

Returns a ProcessorError when a sink serde cannot serialize a record a punctuator forwarded.

pub fn punctuate_wall_clock( &mut self, now_ms: i64, ) -> Result<(), ProcessorError>

Sets the wall clock to now_ms and fires every due wall-clock punctuator once at that time.

The runtime ticks the wall clock between polls; an embedder ticks it from its own clock. A schedule registered during the call stamps its first fire from now_ms.

Errors

Returns a ProcessorError when a sink serde cannot serialize a record a punctuator forwarded.

pub fn flush_caches(&mut self) -> Result<(), ProcessorError>

Flushes every cached store and forwards its deduplicated changes.

The runtime flushes before each commit. The embedded graph runs with the record cache disabled, so no store is cached and the call forwards nothing; it exists so an embedder can follow the runtime’s commit sequence unchanged.

Errors

Returns a ProcessorError when a node rejects a forwarded change.

pub fn clear_stores(&mut self)

Wipes every store, for a clean-slate replay of the changelogs.

pub fn snapshot_stores(&mut self) -> TaskSnapshot

Snapshots every store, keyed by store name, for a barrier cut.

The payloads are the ones SnapshotStore keeps, and restore_store_snapshots takes the task back to them.

pub fn restore_store_snapshots( &mut self, snapshot: &TaskSnapshot, ) -> Result<(), StreamsClientError>

Replaces every store with the snapshot taken at a cut.

A store the snapshot does not name is wiped, so the task holds the state of the cut and nothing after it.

Errors

Returns StreamsClientError::Snapshot when a payload does not match the format its store writes.

pub fn close(self)

Calls close on every processor, in graph order, and drops the task.

pub fn store_names(&self) -> Vec<String>

The names of the task’s state stores, in ascending order.

Global stores are not in the list; they belong to the application, not to a task.

pub fn store_changelog_topic(&self, store: &str) -> Option<String>

The changelog topic of a store, or None when the task has no such store or the store has logging off.

pub fn dump_store(&self, store: &str, limit: usize) -> Vec<(Bytes, Bytes)>

The first limit entries of a key-value store, as raw key and value bytes in ascending key order.

The dump is for an inspector, and it scans the whole store before it cuts the list. A window, session or versioned store dumps nothing here, and neither does a store the task does not hold.