Kafka Streams compatibility matrix
Operator-by-operator parity between the StoatFlow DSL and the Kafka Streams DSL — implemented, partial, not yet implemented, deprecated, incompatible, and StoatFlow extensions.
The engineering view of Kafka Streams DSL parity in StoatFlow: every operator, helper class, and store method, with its current status and notes where behaviour differs. Targets Kafka Streams 4.3.x: StoatFlow reimplements the Kafka Streams DSL surface under its own package, so most KS applications port by swapping the dependency and rewriting imports — see Migrating from Kafka Streams, and let the automated port recipe do it for you.
For the high-level, feature-by-feature comparison against Kafka Streams and Flink, see Comparison matrix. For how-to guidance on the operators below, see Building.
This matrix is transcribed from the in-repository compatibility reference and curated for the public docs. Status reflects the current release (
1.0.0-rc.4). A small number of rows differ in signature from Kafka Streams (marked ⚠️) or are intentionally absent (marked 🛑) — read the notes.Legend
| Symbol | Meaning |
|---|---|
| ✅ | Implemented |
| ⚠️ | Partial — stub, limited, or differing signature |
| ❌ | Not implemented yet |
| 🛑 | Incompatible, or won't be implemented (not needed in the single-instance model) |
| ↘️ | Deprecated in Kafka Streams |
| 🚀 | StoatFlow extension — not in Kafka Streams |
| 🆕 | StoatFlow extension on an otherwise KS-compatible interface |
| 🟡 | Recognised no-op — accepted and warned about, never an error (configuration keys only) |
| 🧩 | Client pass-through — forwarded to the Kafka clients the key is valid for (configuration keys only) |
StreamsBuilder
| Method | Status | Notes |
|---|---|---|
| StreamsBuilder() | ✅ | No-arg |
| StreamsBuilder(TopologyConfig) | ✅ | KSC-77 — seeds build-time defaults (DSL store format, FK reverse-index cache, explicit naming) and the KIP-1112 ProcessorWrapper from a TopologyConfig projection of StreamsConfig. It carries no serde defaults: default.key.serde / default.value.serde on the runtime StreamsConfig decide every serde, as in Kafka Streams. Named-topology / per-topology overrides N/A (single-instance) |
| stream(topic, consumed?) | ✅ | Single topic subscription; single-arg form Java-callable |
| stream(Collection<String>) | ✅ | Multiple topics subscription |
| stream(Pattern) | ✅ | Pattern-based subscription — N patterns combine into one subscription; a topic two patterns accept is rejected at assignment; literals are quoted into the combined regex (KSC-101); no own-topic filtering. Deleting a matched topic is survived (in-place engine restart, ~0.5-1 s, no pod restart), matching KS; a deleted literal source topic stays fatal, also matching KS — see Streams builder |
| table(topic, consumed?) | ✅ | |
| table(topic, materialized) | ✅ | Explicit skip-Consumed overload (KS parity). The Materialized serdes also read the source topic, as in Kafka Streams |
| table(topic, consumed?, materialized) | ✅ | With state store |
| globalTable(topic, consumed?) | ✅ | Wraps KTable |
| globalTable(topic, materialized) | ✅ | Without Consumed parameter. ⚠️ A versioned Materialized is accepted and materialises a versioned store — Kafka Streams throws GlobalTables cannot be versioned. (KSC-125, more permissive) |
| build() | ✅ | Returns Topology |
| build(Properties) | ⚠️ | The Properties are ignored — logs a warning and delegates to build(). Config carries over via StreamsConfig.fromProperties(...) / the KS-shaped StreamsConfig(Properties) ctor instead (ADR-124) |
| addStateStore(storeBuilder) | ✅ | Register standalone store |
| addStateStore(storeBuilder, keySerde, valueSerde) | ✅ | With changelog Serdes |
| addStateStore(storeSupplier, keySerde, valueSerde) | ✅ | Convenience overload |
| addGlobalStore(storeBuilder, topic, consumed, processor) | ✅ | Custom update processor |
| addGlobalStore(storeBuilder, topic, consumed) | ✅ | Default put/delete semantics, written in the store's shape: a timestamped builder holds ValueAndTimestamp(value, record timestamp), a versioned one a version at the record timestamp. 🆕 StoatFlow convenience — Kafka Streams always takes a processor |
| addReadOnlyStateStore(storeBuilder, topic, consumed, processor) | ✅ | KIP-813: custom processor |
| addReadOnlyStateStore(storeBuilder, topic, consumed) | ✅ | KIP-813: Default put/delete semantics, written in the store's shape: a timestamped builder holds ValueAndTimestamp(value, record timestamp), a versioned one a version at the record timestamp. 🆕 StoatFlow convenience — Kafka Streams always takes a processor |
| scheduled(interval, type, emitter) | 🚀 | Interval-based periodic record source |
| scheduled(interval, type, named, emitter) | 🚀 | With explicit naming |
| scheduled(interval, type, keySerde, emitter) | 🚀 | With explicit key serde for lane assignment |
| scheduled(cron, emitter) | 🚀 | Cron-based calendar scheduling |
| scheduled(cron, named, emitter) | 🚀 | Cron with explicit naming |
| scheduled(cron, keySerde, emitter) | 🚀 | Cron with explicit key serde for lane assignment |
Topology (Processor API)
Both the high-level DSL (StreamsBuilder) and the low-level Topology builder are supported.
| Method | Status | Notes |
|---|---|---|
| describe() | ✅ | Returns the Kafka Streams TopologyDescription (nested Source/Processor/Sink/Subtopology/GlobalStore node graph). Breaking vs older StoatFlow: previously returned the native model — that view moved to describeStoatFlow() |
| describeStoatFlow() | 🚀 | StoatFlow-native topology model (the former describe() shape) |
| addSource(name, topics...) | ✅ | 16 Kafka Streams-shaped overloads: offset reset, timestamp extractor, deserializers, pattern (same rules as stream(Pattern)), and 🆕 watermark strategy (no KS equivalent — the Processor API route to a per-source generator, withIdleness and alignment). The deserializers, timestamp extractor and offset reset passed in are used; null means the configured default (KS parity). |
| addSource(name, consumed, topics... | pattern) | 🆕 | Takes a Consumed, mapped exactly as StreamsBuilder.stream() maps it: key/value serdes, timestamp extractor or watermark strategy, offset reset. Unlike a key deserializer, it declares the key serde, so Processor API timers and in-memory sub-topology boundaries behind the source use it instead of default.key.serde. A Consumed.as(...) name must match the node name; withMaterializeFromSourceTopic is refused |
| addProcessor(name, supplier, parents...) | ✅ | User-defined processor with ProcessorSupplier |
| addSink(name, topic, parents...) | ✅ | 8 overloads: serializers, partitioner, TopicNameExtractor. The serializers and partitioner passed in are used; null means the configured default (KS parity). |
| addStateStore(builder, processors...) | ✅ | Register store and optionally connect to processors. Several processors may connect the same builder instance and share the store; a different builder under an existing name is rejected |
| addGlobalStore(...) | ✅ | 2 variants: with/without TimestampExtractor. The deserializers and timestamp extractor passed in are used; null means the configured default (KS parity). |
| addGlobalStore(storeBuilder, sourceName, consumed, topic, processorName, supplier) | 🆕 | The Consumed form, mapped as addSource(name, consumed, …). An offset reset other than earliest is refused: a store's source always reads from the beginning |
| addReadOnlyStateStore(...) | ✅ | 2 variants (KIP-813): source topic is changelog. The deserializers and timestamp extractor passed in are used; null means the configured default (KS parity). |
| addReadOnlyStateStore(storeBuilder, sourceName, consumed, topic, processorName, supplier) | 🆕 | The Consumed form, mapped as addSource(name, consumed, …). An offset reset other than earliest is refused |
| connectProcessorAndStateStores(...) | ✅ | Connect existing store to processor |
TopologyDescription
describe() returns the Kafka Streams TopologyDescription — a graph of nested node interfaces.
| Member | Status | Notes |
|---|---|---|
| subtopologies() / globalStores() | ✅ | Sets of Subtopology / GlobalStore |
| Subtopology.id() / nodes() | ✅ | |
| Node.name() / predecessors() / successors() | ✅ | |
| Source.topicSet() / topicPattern() | ✅ | The matched topics of a pattern source are a runtime fact (stoatflow.source.pattern.matched.topics), not part of the description |
| Processor.stores() | ✅ | Connected store names |
| Sink.topic() / topicNameExtractor() | ✅ | Static or dynamic routing |
| GlobalStore.id() / source() / processor() | ✅ |
KStream — stateless transformations
| Method | Status | Notes |
|---|---|---|
| filter(predicate) | ✅ | |
| filterNot(predicate) | ✅ | |
| map(mapper) | ✅ | Key-changing; the sub-topology boundary opens downstream, where an operator needs the new key's affinity (KS repartitionRequired model, ADR-138) |
| mapValues(ValueMapper) | ✅ | |
| mapValues(ValueMapperWithKey) | ✅ | Key-aware value mapper variant |
| flatMap(mapper) | ✅ | Key-changing; the sub-topology boundary opens downstream, where an operator needs the new key's affinity (KS repartitionRequired model, ADR-138) |
| flatMapValues(ValueMapper) | ✅ | |
| flatMapValues(ValueMapperWithKey) | ✅ | Key-aware value mapper variant |
| selectKey(keySelector) | ✅ | Key-changing; the sub-topology boundary opens downstream, where an operator needs the new key's affinity (KS repartitionRequired model, ADR-138) |
| peek(action) | ✅ | Java BiConsumer overload |
| merge(other) | ✅ | |
| split() | ✅ | Returns BranchedKStream |
| repartition() | ✅ | In-memory only |
| repartition(Repartitioned) | ✅ | Custom partitioner, naming |
| print() | ✅ | Configurable via Printed |
KStream — terminal operations
| Method | Status | Notes |
|---|---|---|
| forEach(action) | ✅ | Java BiConsumer overload |
| to(topic, produced?) | ✅ | Static topic routing |
| to(TopicNameExtractor, produced?) | ✅ | Dynamic topic routing |
| toTable() | ✅ | Auto-generates store name; the store takes the stream's serdes, windowed keys included |
| toTable(named?, materialized?) | ✅ | With naming and materialization; Materialized serdes first, then the stream's (KS parity) |
KStream — Processor API
| Method | Status | Notes |
|---|---|---|
| process(processorSupplier, named?, storeNames...) | ✅ | Key-changing; ProcessorSupplier. Connects the named stores (and any declared via supplier.stores()). Like KS, it does not open a sub-topology boundary of its own — see the note below. A node connected to a store another part of the topology writes is placed with that store's writer, as KS puts processors sharing a store into one subtopology; nodes in different sub-topologies connecting one store warn (presumed-cross-sub-topology-write) — see Sharing a store between processors |
| processValues(processorSupplier, named?, storeNames...) | ✅ | Non-key-changing; FixedKeyProcessorSupplier + named/declared stores. Like KS, it does not open a sub-topology boundary of its own — see the note below. Store placement as for process |
| transform() | 🛑 | Removed from KStream in KS 4.x; use process() |
| transformValues() | 🛑 | Removed from KStream in KS 4.x; use processValues() |
Key affinity at a Processor API node. Kafka Streams never materialises a repartition before
process() / processValues(), even with connected stores, and since 1.0.0 neither does StoatFlow (topology.processor-api-key-affinity: off — presumed restores the pre-1.0.0 boundary). After a many-to-one re-key that means per-key state at such a node is fragmented across lanes, exactly as it is across tasks in KS. StoatFlow adds a guard KS does not have: it refuses to compile the provable case — a proven re-key into a node with a writable store (topology.validation.papi-key-affinity-presumed, default error) — and warns where the feeder is itself a Processor API node, whose key change is only a conservative default (papi-key-affinity-presumed-chain, default warn). See Key affinity after a re-key.KStream — grouping
| Method | Status | Notes |
|---|---|---|
| groupByKey() | ✅ | Returns KGroupedStream |
| groupByKey(Grouped) | ✅ | With Serdes configuration |
| groupBy(keySelector) | ✅ | Key-changing; the sub-topology boundary opens downstream, where an operator needs the new key's affinity (KS repartitionRequired model, ADR-138) |
| groupBy(keySelector, Grouped) | ✅ | With Serdes configuration |
KStream — joins
| Method | Status | Notes |
|---|---|---|
| join(KStream, ValueJoiner, windows) | ✅ | Stream-stream windowed join |
| join(KStream, ValueJoiner, windows, streamJoined) | ✅ | With StreamJoined config |
| join(KStream, ValueJoinerWithKey, windows) | ✅ | Key-aware joiner variant |
| join(KStream, ValueJoinerWithKey, windows, streamJoined) | ✅ | Key-aware joiner with config |
| leftJoin(KStream, ValueJoiner, windows) | ✅ | Stream-stream left join. An unmatched record whose window has already closed when it arrives is emitted at once with a null side, under its own timestamp, never buffered — KS's late branch, judged on the watermark instead of stream time (KSC-132) |
| leftJoin(KStream, ValueJoiner, windows, streamJoined) | ✅ | With StreamJoined config |
| leftJoin(KStream, ValueJoinerWithKey, windows) | ✅ | Key-aware joiner variant |
| leftJoin(KStream, ValueJoinerWithKey, windows, streamJoined) | ✅ | Key-aware joiner with config |
| outerJoin(KStream, ValueJoiner, windows) | ✅ | Stream-stream outer join. Late unmatched records on either side are emitted at once with a null side, as for leftJoin |
| outerJoin(KStream, ValueJoiner, windows, streamJoined) | ✅ | With StreamJoined config |
| outerJoin(KStream, ValueJoinerWithKey, windows) | ✅ | Key-aware joiner variant |
| outerJoin(KStream, ValueJoinerWithKey, windows, streamJoined) | ✅ | Key-aware joiner with config |
| join(KTable, ValueJoiner) | ✅ | Stream-table inner join. After a re-key the join reads the table from another lane: under exactly_once as of the stream record's commit, and an lruMap or in-memory versioned table is refused there (KSC-126, also for leftJoin); with no re-key it reads on its own lane |
| join(KTable, ValueJoinerWithKey) | ✅ | Key-aware joiner variant |
| leftJoin(KTable, ValueJoiner) | ✅ | Stream-table left join |
| leftJoin(KTable, ValueJoinerWithKey) | ✅ | Key-aware joiner variant |
| join(GlobalKTable, keySelector, ValueJoiner) | ✅ | Stream-GlobalKTable inner join. Kotlin: also a (V, GV) -> VR lambda overload, optionally with a Named before the joiner. Under exactly_once every global-table join (join and leftJoin, all overloads) reads the table as of the stream record's commit — stronger than KS, whose global thread gives no cut against the stream tasks; same API. An lruMap or in-memory versioned global table is refused there (KSC-126) |
| join(GlobalKTable, keySelector, ValueJoinerWithKey) | ✅ | Key-aware joiner variant; deprecated since Kafka Streams 4.4 (KIP-1340): the joiner receives the stream key, not the join key — use ValueJoinerWithStreamAndMappedKey |
| join(GlobalKTable, keySelector, ValueJoiner|ValueJoinerWithKey, Named) | ✅ | Named overloads; the ValueJoinerWithKey form is deprecated as above |
| join(GlobalKTable, keySelector, ValueJoinerWithStreamAndMappedKey, Named) | ✅ | KIP-1340 (Kafka Streams 4.4): the joiner receives (streamKey, mappedKey, value1, value2). Kotlin: also a (K, GK, V, GV) -> VR lambda overload, optionally with a Named before the joiner |
| leftJoin(GlobalKTable, keySelector, ValueJoiner) | ✅ | Stream-GlobalKTable left join. Kotlin: also a (V, GV?) -> VR lambda overload, optionally with a Named before the joiner |
| leftJoin(GlobalKTable, keySelector, ValueJoinerWithKey) | ✅ | Key-aware joiner variant; deprecated since Kafka Streams 4.4 (KIP-1340): the joiner receives the stream key, not the join key — use ValueJoinerWithStreamAndMappedKey |
| leftJoin(GlobalKTable, keySelector, ValueJoiner|ValueJoinerWithKey, Named) | ✅ | Named overloads; the ValueJoinerWithKey form is deprecated as above |
| leftJoin(GlobalKTable, keySelector, ValueJoinerWithStreamAndMappedKey, Named) | ✅ | KIP-1340 (Kafka Streams 4.4): the joiner receives (streamKey, mappedKey, value1, value2); mappedKey is null when the selector returns null (the table value is then null too). Kotlin: also a (K, GK?, V, GV?) -> VR lambda overload, optionally with a Named before the joiner |
Temporal join semantics (KIP-914). When the backing table uses a versioned store (KIP-889), stream-table joins perform a temporal lookup against the stream record's timestamp — returning the table value that was valid at the stream event time, so results stay correct when records arrive out of order. Stream-table joins are otherwise point-in-time lookups; a table update does not re-trigger the join.
KStream — branching (BranchedKStream)
| Method | Status | Notes |
|---|---|---|
| branch(predicate) | ✅ | |
| branch(predicate, branched) | ✅ | With Branched config |
| defaultBranch() | ✅ | |
| defaultBranch(branched) | ✅ | With Branched config |
| noDefaultBranch() | ✅ |
KGroupedStream
| Method | Status | Notes |
|---|---|---|
| count() | ✅ | With Materialized overloads |
| reduce(reducer) | ✅ | With Materialized overloads |
| aggregate(initializer, aggregator) | ✅ | With Materialized overloads |
| windowedBy(TimeWindows) | ✅ | Tumbling/hopping windows → TimeWindowedKStream |
| windowedBy(SlidingWindows) | ✅ | Sliding windows (KIP-450) → returns TimeWindowedKStream, exactly like Kafka Streams (no separate SlidingWindowedKStream type) |
| windowedBy(SessionWindows) | ✅ | Session windows |
| cogroup() | ✅ | Returns CogroupedKStream |
Windowed grouped streams
TimeWindowedKStream
Reached by both windowedBy(TimeWindows) (tumbling/hopping) and windowedBy(SlidingWindows) (KIP-450
sliding) — sliding windows flow through this type, exactly as in Kafka Streams. All overloads take a
plain-key Materialized<K, V, …>.
| Method | Status | Notes |
|---|---|---|
| count() | ✅ | With Materialized overloads (plain key) |
| reduce(reducer) | ✅ | With Materialized overloads |
| aggregate(initializer, aggregator) | ✅ | With Materialized overloads |
| emitStrategy() | ✅ | OnWindowUpdate (default), OnWindowClose (lateness and window close follow the watermark, not per-aggregation stream time — KSC-132) |
SessionWindowedKStream
| Method | Status | Notes |
|---|---|---|
| count() | ✅ | With session merger |
| reduce(reducer) | ✅ | Reducer doubles as merger |
| aggregate(initializer, aggregator, sessionMerger) | ✅ | With session merger |
| emitStrategy() | ✅ | OnWindowUpdate (default), OnWindowClose (lateness and window close follow the watermark, not per-aggregation stream time — KSC-132) |
Cogrouped streams
CogroupedKStream
| Method | Status | Notes |
|---|---|---|
| cogroup(KGroupedStream, aggregator) | ✅ | Add another stream to the cogroup |
| aggregate(initializer) | ✅ | With Materialized overloads |
| windowedBy(TimeWindows) | ✅ | Returns TimeWindowedCogroupedKStream |
| windowedBy(SlidingWindows) | ✅ | Sliding windows (KIP-450) → returns TimeWindowedCogroupedKStream |
| windowedBy(SessionWindows) | ✅ | Returns SessionWindowedCogroupedKStream |
TimeWindowedCogroupedKStream
| Method | Status | Notes |
|---|---|---|
| aggregate(initializer) | ✅ | With Materialized overloads |
| emitStrategy() | ✅ | OnWindowUpdate (default), OnWindowClose (lateness and window close follow the watermark, not per-aggregation stream time — KSC-132) |
SessionWindowedCogroupedKStream
| Method | Status | Notes |
|---|---|---|
| aggregate(initializer, sessionMerger) | ✅ | With Materialized overloads; (initializer, sessionMerger) argument order matches Kafka Streams |
| emitStrategy() | ✅ | OnWindowUpdate (default), OnWindowClose (lateness and window close follow the watermark, not per-aggregation stream time — KSC-132) |
KTable — transformations
| Method | Status | Notes |
|---|---|---|
| filter(predicate) | ✅ | Derived table (no state store) |
| filter(predicate, Named) | ✅ | With naming |
| filter(predicate, Materialized) | ✅ | With materialization |
| filter(predicate, Named, Materialized) | ✅ | With naming and materialization |
| filterNot(predicate) | ✅ | Inverse of filter |
| filterNot(predicate, Named) | ✅ | With naming |
| filterNot(predicate, Materialized) | ✅ | With materialization |
| filterNot(predicate, Named, Materialized) | ✅ | With naming and materialization |
| mapValues(ValueMapper) | ✅ | Derived table (no state store) |
| mapValues(ValueMapper, Named) | ✅ | With naming |
| mapValues(ValueMapper, Materialized) | ✅ | With materialization |
| mapValues(ValueMapper, Named, Materialized) | ✅ | With naming and materialization |
| mapValues(ValueMapperWithKey) | ✅ | Key-aware value mapper variant |
| mapValues(ValueMapperWithKey, Named) | ✅ | With naming |
| mapValues(ValueMapperWithKey, Materialized) | ✅ | With materialization |
| mapValues(ValueMapperWithKey, Named, Materialized) | ✅ | With naming and materialization |
| toStream() | ✅ | Same key type |
| toStream(KeyValueMapper) | ✅ | Key-changing variant; the sub-topology boundary opens downstream, where an operator needs the new key's affinity (ADR-138) |
| queryableStoreName() | ✅ | Returns store name if materialized |
| groupBy(keyValueSelector) | ✅ | Selector returns KeyValue<KR, VR>; returns KGroupedTable<KR, VR> |
| groupBy(keyValueSelector, Grouped) | ✅ | With Serdes configuration |
| suppress(Suppressed) | ⚠️ | untilWindowCloses, untilTimeLimit. The buffer falls back to default.key.serde / default.value.serde when no serde was set upstream — including for windowed tables, where Kafka Streams fails (KAFKA-9259). The emitted record currently carries no headers and the watermark as its timestamp — Kafka Streams keeps the buffered record's headers and timestamp. Fix planned for 1.0.0 |
| transformValues(ValueTransformerWithKeySupplier, named?, storeNames...) | ✅ | Stateful per-key value transform; returns KTable<K, VR> |
| transformValues(ValueTransformerWithKeySupplier, Materialized, Named, storeNames...) | ✅ | Materializes the result into a queryable store (tombstone-aware: a null transform deletes). Supplier parameter carries Kafka Streams' ? super K, ? super V, ? extends VR wildcards, so a supplier producing a subtype of the result value type still fits |
KTable — primary-key joins
Both tables share key type K. No co-partitioning required (global state model).
| Method | Status | Notes |
|---|---|---|
| join(KTable, joiner) | ✅ | Inner join; both must be materialized. Each side reads the other from another lane: under exactly_once as of its own commit, and an lruMap or in-memory versioned input table is refused (KSC-126, every primary-key join overload) — a RocksDB versioned one is accepted |
| join(KTable, joiner, Materialized) | ✅ | With materialization of join result |
| join(KTable, joiner, Named, Materialized) | ✅ | With naming and materialization |
| leftJoin(KTable, joiner) | ✅ | Left table always emits |
| leftJoin(KTable, joiner, Materialized) | ✅ | With materialization of join result |
| leftJoin(KTable, joiner, Named, Materialized) | ✅ | With naming and materialization |
| outerJoin(KTable, joiner) | ✅ | Full outer join |
| outerJoin(KTable, joiner, Materialized) | ✅ | With materialization of join result |
| outerJoin(KTable, joiner, Named, Materialized) | ✅ | With naming and materialization |
KIP-914 semantics. When both tables are backed by versioned stores, out-of-order records do not trigger join evaluation, and temporal lookups return the correct value from the other table — preventing inconsistent results under out-of-order arrival.
KTable — foreign-key joins
Foreign-key extractor: a Java Function<V, KO?> (or BiFunction<K, V, KO?> for key-aware extraction) — or a Kotlin (V) -> KO? / (K, V) -> KO? lambda.
| Method | Status | Notes |
|---|---|---|
| join(KTable, fkExtractor, joiner) | ✅ | Inner FK join. ⚠️ An lruMap or in-memory versioned input table on either side is refused under exactly_once (KSC-126, all FK join overloads); a RocksDB versioned one is accepted |
| join(KTable, fkExtractor, joiner, Materialized) | ✅ | With materialization of join result |
| join(KTable, fkExtractor, joiner, TableJoined, Materialized) | ✅ | With naming and materialization |
| leftJoin(KTable, fkExtractor, joiner) | ✅ | Left FK join; a null FK emits joiner(left, null) (KIP-962). FK-join retractions are forwarded materialized or not, as in Kafka Streams |
| leftJoin(KTable, fkExtractor, joiner, Materialized) | ✅ | With materialization of join result |
| leftJoin(KTable, fkExtractor, joiner, TableJoined, Materialized) | ✅ | With naming and materialization |
| outerJoin(KTable, fkExtractor, joiner) | ⚠️ | Not supported in KS either |
KGroupedTable
KGroupedTable is the result of KTable.groupBy() and uses changelog semantics — moving a key from group A to group B decrements A and increments B. The source KTable must be materialized for groupBy() to track old values.
| Method | Status | Notes |
|---|---|---|
| count() | ✅ | Changelog semantics |
| reduce(adder, subtractor) | ✅ | Kafka Streams compatible; both adder and subtractor are Reducer<V> |
| aggregate(init, adder, subtractor) | ✅ | Kafka Streams compatible; both adder and subtractor are Aggregator<K, V, VR> — no StoatFlow-only Subtractor type |
GlobalKTable
| Method | Status | Notes |
|---|---|---|
| queryableStoreName() | ✅ | Returns underlying table's store name |
| asKTable() | 🚀 | StoatFlow extension to unwrap |
GlobalKTable wraps KTable internally. Because all state is global, its behaviour is identical to KTable.
Processor API
ProcessorContext
| Method | Status | Notes |
|---|---|---|
| forward(record) | ✅ | Forward to downstream processors |
| forward(record, childName) | ✅ | Forward to specific child |
| getStateStore(name) | ✅ | Access connected state stores — declared via ProcessorSupplier.stores() or the process(supplier, "name") varargs. An undeclared store throws StoreNotConnectedException (KS parity). Global stores (addGlobalStore, globalTable) are exempt and returned read-only unless the node declares them, in which case they are writable (a StoatFlow extension — KS makes declaring a global store a TopologyException). The read-only view covers every store family (plain, timestamped, versioned, headers-aware) and keeps the store's declared type, so the caller's getStateStore<TimestampedKeyValueStore<..>> cast still succeeds; read-only stores (addReadOnlyStateStore) are exempt too — a deliberate divergence, KS exempts only globals |
| schedule(interval, type, punctuator) | ✅ | KS-compatible punctuator, KS fire times (STREAM_TIME aligned to 0, a grid, one fire per catch-up); inside the callback context.timestamp() is the punctuation time and headers() empty, as in KS |
| schedule(interval, type, mode, punctuator) | 🚀 | StoatFlow extension with BLOCKING/NON_BLOCKING mode |
| schedule(interval, startTime, type, punctuator) | ✅ | KIP-1146: anchored punctuation with grid-aligned fire times |
| schedule(interval, startTime, type, mode, punctuator) | ✅ | KIP-1146: anchored punctuation with BLOCKING/NON_BLOCKING mode |
| recordMetadata() | ✅ | Source partition, offset, topic |
| applicationId() | ✅ | The configured application.id, on every context; under the TopologyTestDriver, the driver config's id |
| currentSystemTimeMs() | ✅ | Wall-clock time via System.currentTimeMillis(); under the TopologyTestDriver, the driver's wall clock, as in Kafka Streams |
| currentStreamTimeMs() | ✅ | Delegates to currentWatermarkMs() (global watermark) |
| currentWatermarkMs() | 🆕 | Global watermark in epoch millis |
| timerService() | 🚀 | Key-based timer access (Flink-inspired) |
Processor interfaces
| Type | Status | Notes |
|---|---|---|
| Processor<KIn,VIn,KOut,VOut> | ✅ | Key-changing processor interface |
| FixedKeyProcessor<KIn,VIn,VOut> | ✅ | Key-preserving processor interface |
| ContextualProcessor<KIn,VIn,KOut,VOut> | ✅ | Abstract base managing ProcessorContext |
| ContextualFixedKeyProcessor<KIn,VIn,VOut> | ✅ | Abstract base managing FixedKeyProcessorContext |
| ProcessorSupplier<KIn,VIn,KOut,VOut> | ✅ | Factory for Processor instances |
| ProcessorSupplier.stores() | ✅ | KS 4.x pattern for declaring stores inline |
| FixedKeyProcessorSupplier<KIn,VIn,VOut> | ✅ | Factory for FixedKeyProcessor instances |
| ValueTransformerWithKey<K,V,VR> | ✅ | Consumed by KTable.transformValues; init binds StoatFlow's FixedKeyProcessorContext |
| ValueTransformerWithKeySupplier<K,V,VR> | ✅ | Factory for ValueTransformerWithKey; declares stores via stores(). As a KTable.transformValues parameter it carries Kafka Streams' ? super K, ? super V, ? extends VR wildcards |
| ValueTransformer / Transformer (+ suppliers) | ↘️ | Deprecated in KS and removed from KStream; kept as nameable types for source compatibility |
init() / close() lifecycle | ✅ | Once per task — init() at task start, before any records (application start, an HA pod's activation, TopologyTestDriver construction), close() once at shutdown (KS contract), on every Processor API route (process / processValues / transformValues / addProcessor). A punctuator scheduled in init() starts without input. A global or read-only store's update processor follows the route that added it: through Topology.addGlobalStore / addReadOnlyStateStore it initialises at task start, like addProcessor; through StreamsBuilder.addGlobalStore / addReadOnlyStateStore it initialises on each lane's first record |
FixedKeyProcessor.onTimer + schedule() / timerService() | ✅ | Full Processor timer parity for FixedKeyProcessor (processValues): periodic punctuators via schedule() and keyed timers via timerService() → onTimer(...) (key-preserving FixedKeyTimerContext, full state access, re-registration). Record timestamp is the event time |
Punctuators and timers
| Type / Method | Status | Notes |
|---|---|---|
| PunctuationType.STREAM_TIME | ✅ | Fires on event-time watermark advancement |
| PunctuationType.WALL_CLOCK_TIME | ✅ | Fires on system clock |
| PunctuatorMode.BLOCKING | 🚀 | Barriers wait for punctuator completion before commit |
| PunctuatorMode.NON_BLOCKING | 🚀 | Barriers proceed independently (default) |
| Punctuator.punctuate(timestamp) | ✅ | Callback invoked at scheduled time |
| Cancellable.cancel() | ✅ | Cancel scheduled punctuator or timer |
| TimerService.registerEventTimeTimer(key, timestamp) | 🚀 | Fire when watermark ≥ timestamp |
| TimerService.registerProcessingTimeTimer(key, timestamp) | 🚀 | Fire at wall-clock time |
| TimerService.deleteEventTimeTimer(key, timestamp) | 🚀 | Cancel event-time timer |
| TimerService.deleteProcessingTimeTimer(key, timestamp) | 🚀 | Cancel processing-time timer |
| TimerService.currentWatermark() | 🚀 | Current event-time watermark |
| TimerService.currentProcessingTime() | 🚀 | Current wall-clock time (the driver's wall clock under the TopologyTestDriver) |
Kafka Streams punctuators are partition-scoped with writable state and always non-blocking. StoatFlow punctuators run globally (topology-wide state view) on a dedicated punctuation lane per sub-topology, with full read/write state access; a punctuation lane writes only after every record lane of its sub-topology has crossed the commit barrier, so its writes and the record lanes' are never reordered across a commit, under either guarantee.
PunctuatorMode and the key-based TimerService (which fires in the key's affinity lane, also with full read/write state access) are StoatFlow extensions.Scheduled sources (StoatFlow extension)
A ScheduledEmitter — registered via StreamsBuilder.scheduled(...) — periodically emits records (interval- or cron-based) using the ScheduledEmitterContext below, without consuming from Kafka.
ScheduledEmitterContext
| Method | Status | Notes |
|---|---|---|
| forward(key, value) | 🚀 | Emit record with current wall-clock time |
| forward(key, value, timestamp) | 🚀 | Emit record with explicit timestamp |
| forward(record) | 🚀 | Emit record with key, value, timestamp, headers |
| getStateStore(name) | 🚀 | Access a state store read-only — writes throw UnsupportedOperationException. Every family is covered (plain, timestamped, versioned, headers-aware) and the view keeps the store's declared type, so a getStateStore<TimestampedKeyValueStore<..>> cast still succeeds |
| currentWatermarkMs() | 🆕 | Current watermark (global event-time progress) |
| currentStreamTimeMs() | ✅ | Delegates to currentWatermarkMs() (KS compat) |
| currentSystemTimeMs() | ✅ | Wall-clock time via System.currentTimeMillis() (KS compat); under the TopologyTestDriver, the driver's wall clock, the same as currentWallClockTime() |
| currentWallClockTime() | 🚀 | Current wall-clock time |
CronExpression
| Method | Status | Notes |
|---|---|---|
| unix(expression) | 🚀 | Unix cron (5 fields: min hour dom mon dow) |
| quartz(expression) | 🚀 | Quartz cron (6 fields with seconds) |
| spring(expression) | 🚀 | Spring cron (6 fields with seconds) |
| parse(expression, cronType) | 🚀 | Parse with explicit type |
| nextExecutionTime(afterMs, zoneId?) | 🚀 | Calculate next fire time |
| timeUntilNextExecution(fromMs, zoneId?) | 🚀 | Duration until next fire |
Cron-based scheduling always uses wall-clock time.
Functional interfaces
KS-compatible functional interfaces backing the Java-friendly DSL overloads.
| Interface | Status | Used in |
|---|---|---|
KeyValueMapper<K, V, VR> | ✅ | map, flatMap, groupBy, FK joins |
ValueMapper<V, VR> | ✅ | KTable.mapValues, FK join extractors |
ValueMapperWithKey<K, V, VR> | ✅ | mapValues with key access |
Predicate<K, V> | ✅ | filter, filterNot |
ForeachAction<K, V> | ✅ | peek, forEach |
Initializer<VA> | ✅ | All aggregate() |
Aggregator<K, V, VA> | ✅ | All aggregate() — incl. both the adder and subtractor of KGroupedTable.aggregate() (KS uses Aggregator for both; no Subtractor type) |
Reducer<V> | ✅ | All reduce() |
Merger<K, V> | ✅ | Session window aggregate() |
ValueJoiner<V1, V2, VR> | ✅ | All joins |
ValueJoinerWithKey<K, V1, V2, VR> | ✅ | Key-aware joins. In a GlobalKTable join the key is the stream key, not the join key — those overloads are deprecated since Kafka Streams 4.4 |
ValueJoinerWithStreamAndMappedKey<StreamKey, TableKey, StreamValue, TableValue, VOut> | ✅ | Kafka Streams 4.4 (KIP-1340): KStream–GlobalKTable joins, apply(streamKey, mappedKey, value1, value2). On a left join the mapped key and the table value may be null, so a Kotlin left-join joiner types them nullable — one declared with non-null type arguments compiles on Kafka Streams (Java platform types) but not here. A null-value record is dropped before the key selector runs (Kafka Streams runs it first) |
StreamPartitioner<K, V> | ✅ | Kafka Streams 4.4 shape (KIP-1321): the default partitions(topic, key, value, headers, numPartitions) is what StoatFlow calls, with the record's headers; the KIP-837 4-arg partitions(...) stays the single abstract method, deprecated for removal. Sink multicast matches KS; the in-memory repartition path throws on multicast (single-instance) and passes the lane count. Lives in the topology package |
Configuration helper classes
Consumed
| Method | Status | Notes |
|---|---|---|
| with(keySerde, valueSerde) | ✅ | |
| with(keySerde, valueSerde, timestampExtractor, resetPolicy) | ✅ | Extractor adapted to a stream-time watermark strategy |
| keySerde(serde) | ✅ | |
| valueSerde(serde) | ✅ | |
| as(name) | ✅ | |
| withWatermarkStrategy(strategy) | 🚀 | StoatFlow extension (mutually exclusive with withTimestampExtractor) |
| withTimestampExtractor() | ✅ | KS TimestampExtractor adapted to a per-source watermark strategy that assigns event time + emits stream-time watermarks (windowing/suppress close as in KS); use withWatermarkStrategy for bounded-out-of-orderness/idleness |
| withOffsetResetPolicy() | ✅ | Per-source offset reset. AutoOffsetReset: Earliest / Latest / None / byDuration(Duration) (KIP-1106 — seek to the first offset at/after now − duration) |
| withNumberOfLanes(int) | 🆕 | Sets this source chain's lane count (parallelism) |
| withLaneQueueCapacity(int) | 🆕 | Sets this source chain's per-lane queue capacity |
WatermarkStrategy (Flink-inspired)
Event-time progress is driven by a WatermarkStrategy. A KS Consumed.withTimestampExtractor(...) is supported and adapted to one under the hood; Consumed.withWatermarkStrategy(...) is the richer StoatFlow-native form (bounded out-of-orderness, idleness, alignment).
| Method | Status | Notes |
|---|---|---|
| forBoundedOutOfOrderness(maxOutOfOrderness) | 🚀 | Bounded out-of-orderness |
| forMonotonousTimestamps() | 🚀 | Strictly increasing timestamps |
| noWatermarks() | 🚀 | Processing-time only |
| withTimestampAssigner(assigner) | 🚀 | Custom event-time assignment |
| withIdleness(duration) | 🚀 | Idle-source handling |
| withWatermarkAlignment(group, maxDrift, updateInterval) | 🚀 | Flink FLIP-217 alignment, Flink's signature. A partition whose watermark drifts beyond the group cap is held, never paused: it keeps being fetched, its records go only in timestamp order, and while a lagging partition has nothing buffered the whole application waits (the drift bound applies against empty laggards; strict order otherwise). Requires idleness on the group's sources. See Watermark alignment |
The TimestampExtractor SAM is present with the KS-exact, non-generic signature long extract(ConsumerRecord<Object, Object>, long), and Consumed.withTimestampExtractor(...) wires it to event-time + stream-time watermarks.
KS-optional builder arguments (serdes, grace, partitioner, offset-reset, name) across Consumed/Grouped/Produced/Repartitioned/Joined/StreamJoined/Materialized/Branched accept null (= "use the default"), matching Kafka Streams; Named itself still requires a non-null name.
Grouped
| Method | Status | Notes |
|---|---|---|
| as(name) | ✅ | |
| keySerde(serde) | ✅ | |
| valueSerde(serde) | ✅ | |
| with(keySerde, valueSerde) | ✅ | |
| with(name, keySerde, valueSerde) | ✅ | |
| withName(name) | ✅ | |
| withKeySerde(serde) | ✅ | |
| withValueSerde(serde) | ✅ | |
| withNumberOfLanes(int) | 🆕 | Sets the grouped sub-topology's lane count |
| withLaneQueueCapacity(int) | 🆕 | Sets the grouped sub-topology's per-lane queue capacity |
Produced
| Method | Status | Notes |
|---|---|---|
| with(keySerde, valueSerde) | ✅ | |
| keySerde(serde) | ✅ | |
| valueSerde(serde) | ✅ | |
| as(name) | ✅ | |
| streamPartitioner(partitioner) | ✅ |
Materialized
| Method | Status | Notes |
|---|---|---|
| as(storeName) | ✅ | |
| as(storeSupplier) | ✅ | |
| as(DslStoreSuppliers) | ✅ | Store type supplier. Applies only when no explicit supplier is set — an explicit as(supplier) wins (Kafka Streams' order since KSC-100, 2026-09-09) |
| with(keySerde, valueSerde) | ✅ | |
| keySerde(serde) | ✅ | |
| valueSerde(serde) | ✅ | |
| withCachingEnabled() | ✅ | Sets the store's cache caps (default caps, or EmissionSuppression limits). Caching is always on and never changes downstream emissions — every update is forwarded (KSC-88; consolidate with Suppressed.untilBarrier()) |
| withCachingDisabled() | ✅ | Accepted, changes nothing: caching cannot be turned off and the store keeps the default caps; every update is forwarded anyway (KSC-88) |
| withLoggingEnabled() | ✅ | Enable changelog with optional topic config |
| withLoggingDisabled() | ✅ | Disable changelog for this store |
| withRetention() | ✅ | Override computed retention for windowed stores. It is a minimum — a persistent store keeps up to retention + segmentInterval; see window and session retention |
| withStoreType(DslStoreSuppliers) | ✅ | Store type selection. Keeps an explicit supplier, which wins over it (Kafka Streams' order since KSC-100); both set → one WARN per store at build |
| withRecordHeaders() | 🆕 | Persist the record's headers alongside key/value in the store (KIP-1271 / KIP-1285) |
| withoutRecordHeaders() | 🆕 | Disable header persistence (default) |
| withChangelogNumPartitions(n) | 🆕 | Per-store changelog partition count. No KS equivalent — KS co-partitioning forces the count to equal the source topic's, so it is never a decision there. Create-time immutable, and bidirectional: with the default at 3, pinning a small store back to 1 is the common use. On a foreign-key join the internal -subscription-store inherits it too |
Branched
| Method | Status | Notes |
|---|---|---|
| as(name) | ✅ | |
| withName(name) | ✅ | |
| withFunction(chain) | ✅ | |
| withConsumer(consumer) | ✅ | Java Consumer overload |
Named
Named is open/subclassable, so KSML's name-validation subclasses work.
| Method | Status | Notes |
|---|---|---|
| as(name) | ✅ | |
| withName(name) | ✅ | Fluent builder form |
| name() | ✅ | Read the configured name |
Printed
| Method | Status | Notes |
|---|---|---|
| toSysOut() | ✅ | Default stdout |
| toFile(filePath) | ✅ | File output |
| withLabel(label) | ✅ | Prefix label |
| withKeyValueMapper(mapper) | ✅ | Custom formatting; takes a KS KeyValueMapper<K, V, String> |
| withName(processorName) | ✅ | Set processor name |
Joined
Joined<K, VLeft, VRight> is the Kafka Streams generic form. Its serdes and grace period are advisory — stored and round-tripped for source compatibility, but StoatFlow resolves the effective serdes/grace from the upstream Consumed/Grouped configuration.
| Method | Status | Notes |
|---|---|---|
| as(name) | ✅ | |
| create() | ✅ | Create with defaults |
| with(keySerde, valueSerde, otherValueSerde) | ✅ | Static factory; serdes advisory |
| withName(name) | ✅ | |
| withGracePeriod(duration) | ✅ | Advisory grace period |
| withKeySerde(serde) | ✅ | Advisory |
| withValueSerde(serde) | ✅ | Advisory |
| withOtherValueSerde(serde) | ✅ | Advisory |
| name() | ✅ | Read the configured name |
| gracePeriod() | ✅ | Round-trips the advisory grace period |
| keySerde() / valueSerde() / otherValueSerde() | ✅ | Round-trip the advisory serdes |
TableJoined
| Method | Status | Notes |
|---|---|---|
| as(name) | ✅ | |
| with() | ✅ | Defaults |
| with(partitioner, otherPartitioner) | ✅ | Accepted no-op stub (partitioning is N/A single-instance) |
| withName(name) | ✅ | |
| withPartitioner() | ✅ | Accepted no-op stub for KS source compatibility |
| withOtherPartitioner() | ✅ | Accepted no-op stub for KS source compatibility |
JoinWindows
| Method | Status | Notes |
|---|---|---|
| of(timeDifference) | ↘️ | Deprecated; use ofTimeDifferenceWithNoGrace |
| ofTimeDifferenceWithNoGrace(timeDifference) | ✅ | Symmetric window, no grace |
| ofTimeDifferenceAndGrace(timeDifference, grace) | ✅ | Symmetric window with grace |
| before(timeDifference) | ✅ | Asymmetric window (look back) |
| after(timeDifference) | ✅ | Asymmetric window (look forward) |
| grace(duration) | ↘️ | Deprecated; use ofTimeDifferenceAndGrace |
| size() | ✅ | Window size in ms |
| gracePeriodMs() | ✅ | Grace period in ms |
beforeMs / afterMs are exposed as read-only properties (getBeforeMs() / getAfterMs() from Java; before / after return Duration). A KS port that reads the public jw.beforeMs field replaces it with the accessor.StreamJoined
| Method | Status | Notes |
|---|---|---|
| as(name) | ✅ | Create with processor name |
| with(keySerde, leftValueSerde, rightValueSerde) | ✅ | Create with Serdes |
| create() | ✅ | Create with defaults |
| withName(name) | ✅ | Set processor name |
| withKeySerde(serde) | ✅ | Set key Serde |
| withLeftValueSerde(serde) | ✅ | Set left stream value Serde |
| withRightValueSerde(serde) | ✅ | Set right stream value Serde |
| withThisStoreSupplier(storeSupplier) | ✅ | Custom left window store. Must retain duplicates — a non-duplicate supplier is rejected at DSL build (KSC-99), where Kafka Streams accepts it and silently drops same-(key, timestamp) matches. Wins over withDslStoreSuppliers for its side and no longer clears it (Kafka Streams' order since KSC-100) |
| withOtherStoreSupplier(storeSupplier) | ✅ | Custom right window store; same KSC-99 guard, same KSC-100 precedence |
| withLoggingEnabled() | ✅ | Enable changelog logging |
| withLoggingEnabled(config) | ✅ | Enable changelog logging with topic config |
| withLoggingDisabled() | ✅ | Disable changelog logging |
| withDslStoreSuppliers(suppliers) | ✅ | Each join store without an explicit supplier uses this type — an explicit supplier wins for its side and neither setter clears the other (Kafka Streams' order since KSC-100, 2026-09-09). Both retain duplicates (fixed 2026-09-03 — before that this path silently built a non-duplicate join store); a custom DslStoreSuppliers that ignores DslWindowParams.retainDuplicates is rejected at DSL build (KSC-99) |
| withStoreName(name) | ✅ | Set base store name |
| withChangelogNumPartitions(n) | 🆕 | Changelog partition count for this join's stores. No KS equivalent. The number is multiplied — one call sets it on 2 stores (INNER) or 3 (LEFT/OUTER, adding -outer-store), so (6) is 12 or 18 partitions |
Repartitioned
| Method | Status | Notes |
|---|---|---|
| as(name) | ✅ | |
| streamPartitioner(partitioner) | ✅ | numPartitions = numLanes |
| withStreamPartitioner(partitioner) | ✅ | |
| withName(name) | ✅ | |
| with(keySerde, valueSerde) | ✅ | keySerde used for lane assignment hashing |
| withKeySerde() | ✅ | Used for lane assignment hashing at sub-topology boundary |
| withValueSerde() | ✅ | Reserved for future use |
| withNumberOfLanes(int) | 🆕 | Sets the repartition sub-topology's lane count |
| numberOfLanes(int) | 🆕 | Static factory; sets the lane count |
| withLaneQueueCapacity(int) | 🆕 | Sets the per-lane queue capacity |
| withNumberOfPartitions() | 🆕 | KS-compat alias of withNumberOfLanes (sets the lane count) |
| numberOfPartitions(int) | 🆕 | KS-compat static-factory alias of numberOfLanes |
EmitStrategy
| Method | Status | Notes |
|---|---|---|
| onWindowUpdate() | ✅ | Default; emit on every update |
| onWindowClose() | ✅ | Emit only when window closes; driven by the global event-time watermark |
OnWindowClose emission is watermark-driven: a window's final result is emitted when the global watermark passes the window end plus any configured grace period. See Event time and watermarks.Suppressed
| Method | Status | Notes |
|---|---|---|
| untilWindowCloses(bufferConfig) | ✅ | For windowed KTables; requires a StrictBufferConfig (KS parity); flush driven by the global event-time watermark. The grace period comes from the window definition upstream, through any operators in between (the largest across a merge); with no windowed ancestor the build fails with Kafka Streams' "Window close time is only defined for windowed computations" |
| untilTimeLimit(duration, bufferConfig) | ✅ | Rate-limits per-key updates; flush driven by the global event-time watermark |
| withName(name) | ✅ | KS NamedOperation; names the suppression node |
Suppression buffers are flushed on watermark advancement —
untilWindowCloses releases a key's final value once the window closes, and untilTimeLimit releases the latest buffered value once the per-key time limit elapses in event time. See Event time and watermarks. Suppress buffers (and versioned stores, which the migration tool supports as experimental) are the two store families with state-carrying migration caveats.KIP-914:
suppress() throws a validation exception for tables backed by versioned stores — their temporal semantics conflict with suppression.BufferConfig
| Method | Status | Notes |
|---|---|---|
| unbounded() | ✅ | No limits (use with caution) |
| maxRecords(count) | ✅ | Bounded by record count |
| maxBytes(bytes) | ⚠️ | Accepted so KS code compiles, but throws at build time — in-memory object buffers have no byte size |
| withMaxRecords(count) | ✅ | Fluent record bound |
| withMaxBytes(bytes) | ⚠️ | Same as maxBytes — throws |
| withNoBound() | ✅ | Remove bounds |
| emitEarlyWhenFull() | ✅ | Emit instead of evict on overflow |
| shutDownWhenFull() | ✅ | Shut down on buffer overflow |
| withLoggingEnabled(config) / withLoggingDisabled() | ✅ | Changelog control for the suppress buffer |
| withChangelogNumPartitions(n) / changelogNumPartitions() | 🆕 | Changelog partition count for the buffer store. No KS equivalent; returns BC, preserving the self-type. The buffer follows the global changelog.num-partitions by default, like every other store; set this only on a measurement |
WindowedSerdes
| Method | Status | Notes |
|---|---|---|
| timeWindowedSerdeFrom(...) | ⚠️ | Takes Serde<T> instead of Class<T> (no serde registry) |
| timeWindowedSerdeFrom(..., windowSize) | ⚠️ | Takes Serde<T> only; window size encoded in serialized bytes |
| sessionWindowedSerdeFrom(...) | ⚠️ | Takes Serde<T> instead of Class<T> |
| TimeWindowedSerde(inner) | ✅ | WindowedSerdes.TimeWindowedSerde(serde) |
| TimeWindowedSerde(inner, windowSize) | ⚠️ | Window size not required (encoded in serialized bytes) |
| SessionWindowedSerde(inner) | ✅ | WindowedSerdes.SessionWindowedSerde(serde) |
StoatFlow's factory methods take
Serde<T> rather than KS's Class<T> (there is no serde registry), and omit the window-size parameter (window bounds are encoded in the serialized bytes). TimeWindowedSerde.forChangelog() is not needed — StoatFlow uses identical encoding for client and changelog.State stores
Record types & iterators
| Type | Status | Notes |
|---|---|---|
| Windowed<K> | ✅ | Windowed-key wrapper (key + Window) |
| ValueAndTimestamp<V> | ✅ | Timestamped-store value wrapper (KIP-258). Null semantics as in Kafka Streams: make(null, ts) returns null, makeAllowNullable an instance; a null value is written as a delete and a null-deserialized read returns null |
| VersionedRecord<V> | ✅ | Versioned-store record; validTo() returns Optional<Long> |
| KeyValueIterator / WindowStoreIterator | ✅ | Closeable iterators for range/scan queries; use after close() throws InvalidStateStoreException, as in Kafka Streams — and so does any read on a closed store |
State store metrics
| Metric | Status | Notes |
|---|---|---|
| num-keys (KIP-1250) | ✅ | Entry-count gauge for in-memory KV/window/session stores |
KeyValueStore
| Method | Status | Notes |
|---|---|---|
| get(key) | ✅ | |
| put(key, value) | ✅ | |
| putIfAbsent(key, value) | ✅ | |
| putAll(entries) | ✅ | |
| delete(key) | ✅ | Reads and returns the previous value in both engines — a read-then-write. Prefer put(key, null) when you do not need it: Deleting a key |
| compute(key, remappingFunction) | 🚀 | Atomic read-compute-write; Kotlin lambda + Java BiFunction overloads |
| merge(key, value, remappingFunction) | 🚀 | Atomic merge; Kotlin lambda + Java BiFunction overloads |
| all() | ✅ | Read-your-writes semantics |
| range(from, to) | ✅ | Read-your-writes semantics. Null bounds = open-ended (KIP-763, KSC-89); from > to returns empty + WARN (KS-lenient) |
| reverseAll() | ✅ | Read-your-writes semantics |
| reverseRange(from, to) | ✅ | Read-your-writes semantics. Null bounds = open-ended (KIP-763, KSC-89); from > to returns empty + WARN (KS-lenient); (low, high) bounds order, KS-exact (KSC-90 — was (high, low) pre-2026-07-31) |
| approximateNumEntries() | ✅ | |
| prefixScan() | ✅ | KIP-614; read-your-writes semantics |
ReadOnlyKeyValueStore
| Method | Status | Notes |
|---|---|---|
| get(key) | ✅ | |
| containsKey(key) | 🚀 | StoatFlow extension; bloom filter optimization for RocksDB |
| all() | ✅ | |
| range(from, to) | ✅ | Null bounds = open-ended (KIP-763, KSC-89); from > to returns empty + WARN (KS-lenient) |
| reverseAll() | ✅ | |
| reverseRange(from, to) | ✅ | Null bounds = open-ended (KIP-763, KSC-89); from > to returns empty + WARN (KS-lenient); (low, high) bounds order, KS-exact (KSC-90 — was (high, low) pre-2026-07-31) |
| approximateNumEntries() | ✅ | |
| prefixScan(prefix, serializer) | ✅ | KIP-614 |
WindowStore
Key-range bounds are compared as serialized bytes (unsigned lexicographic); null bounds are open-ended (KIP-763, KSC-91). All Instant variants delegate to the Long variants.
| Method | Status | Notes |
|---|---|---|
| put(key, value, windowStartTime) | ✅ | |
| fetch(key, windowStartTime) | ✅ | Point lookup |
| fetch(key, timeFrom, timeTo) | ✅ | Single key time range; Long and Instant variants |
| backwardFetch(key, timeFrom, timeTo) | ✅ | Reverse iteration; Long and Instant variants |
| fetch(keyFrom, keyTo, timeFrom, timeTo) | ✅ | Key range + time range; Long and Instant variants. Null key bounds = open-ended (KIP-763, KSC-91); keyFrom > keyTo returns empty + WARN (KS-lenient) |
| backwardFetch(keyFrom, keyTo, timeFrom, timeTo) | ✅ | Reverse key range; bounds stay (low, high) like fetch (KS-exact); Long and Instant variants. Null key bounds = open-ended (KIP-763, KSC-91) |
| fetchAll(timeFrom, timeTo) | ✅ | All keys time range; Long and Instant variants |
| backwardFetchAll(timeFrom, timeTo) | ✅ | Reverse all keys; Long and Instant variants |
| all() | ✅ | All entries |
| backwardAll() | ✅ | Reverse iteration |
| windowSizeMs | 🚀 | StoatFlow extension for window size access |
| retentionMs | 🚀 | StoatFlow extension for retention access |
| expireWindows(watermark) | 🚀 | StoatFlow extension for manual expiration |
| approximateNumEntries() | 🚀 | StoatFlow extension for entry count |
ReadOnlyWindowStore
| Method | Status | Notes |
|---|---|---|
| All query methods from WindowStore | ✅ | 16 query methods total |
| containsKey(key, windowStartTime) | 🚀 | StoatFlow extension; Long and Instant variants |
SessionStore
| Method | Status | Notes |
|---|---|---|
| put(sessionKey, aggregate) | ✅ | Windowed key |
| remove(sessionKey) | ✅ | Windowed key |
| fetchSession(key, startTime, endTime) | ✅ | Long and Instant variants |
| findSessions(key, earliestEndTime, latestStartTime) | ✅ | Long and Instant variants |
| backwardFindSessions(key, earliestEndTime, latestStartTime) | ✅ | Long and Instant variants |
| findSessions(keyFrom, keyTo, earliestEndTime, latestStartTime) | ✅ | Key range; Long and Instant variants. Null key bounds = open-ended (KIP-763, KSC-91); keyFrom > keyTo returns empty + WARN (KS-lenient) |
| backwardFindSessions(keyFrom, keyTo, earliestEndTime, latestStartTime) | ✅ | Key range; Long and Instant variants. Null key bounds = open-ended (KIP-763, KSC-91) |
| fetch(key) | ✅ | All sessions for single key |
| backwardFetch(key) | ✅ | Reverse order |
| fetch(keyFrom, keyTo) | ✅ | Key range queries. Null key bounds = open-ended (KIP-763, KSC-91); keyFrom > keyTo returns empty + WARN (KS-lenient) |
| backwardFetch(keyFrom, keyTo) | ✅ | Reverse order; bounds stay (low, high) (KS-exact). Null key bounds = open-ended (KIP-763, KSC-91) |
| findAll() | 🚀 | StoatFlow extension |
| expireSessions(watermark) | 🚀 | StoatFlow extension |
| approximateNumEntries() | 🚀 | StoatFlow extension |
ReadOnlySessionStore
| Method | Status | Notes |
|---|---|---|
| All query methods from SessionStore | ✅ | Query methods for IQ access |
| containsSession(key, startTime, endTime) | 🚀 | StoatFlow extension; Long and Instant variants |
Stores factory
StoatFlow's suppliers implement the KS *BytesStoreSupplier interfaces and the *StoreBuilder(...) factories return a KS StoreBuilder<T>. Some signatures take a Serde<T> where KS takes Class<T> (no serde registry); StoreBuilder.withCaching*/withLogging* are honored, and build() throws (StoatFlow builds stores via the registry). A custom StateStoreBuilder whose build() returns a persistent supplier's get() is refused at startup (KSC-124): register the supplier through Stores.keyValueStoreBuilder(supplier, …) (or the window/session equivalent) or Materialized.as(supplier).
| Method | Status | Notes |
|---|---|---|
| persistentKeyValueStore(name) | ✅ | RocksDB key-value store |
| inMemoryKeyValueStore(name) | ✅ | In-memory key-value store |
| persistentWindowStore(name, retention, windowSize, retainDuplicates) | ✅ | RocksDB window store. Segments are sized max(retention / 3, 60s, windowSize) (KS: max(retention / 2, 60s)), with no way to configure it, and expiry drops whole segments — so effective retention is up to 4/3× what you configure where the divisor governs (retention ≥ 180s), up to retention + 60s where the 60s floor governs (see the windowing docs), and up to 2× where windowSize governs on the segmented backend, which is the common case since ofSizeWithNoGrace makes retention == windowSize (retention is a minimum bound, as in Kafka Streams, whose own slack is up to 3/2×; in-memory stores retain exactly 1×). Changing retention on an existing segmented store is safe — each segment records its own time range on disk, so the change remaps rather than being reinterpreted; Kafka Streams requires a wipe. Only a hand-built non-segmented store (no Stores factory builds one) keeps the interval in every key, where a retention change that moves it needs a restore. Changing the window size, switching plain↔timestamped, or flipping retainDuplicates is detected at startup and answered with a full restore from the changelog — or, for a store with no changelog, a refusal to start (stoatflow.state.layout-mismatch). retainDuplicates=true (the stream-stream join-buffer setting) uses the same segmented backend since 2026-09-03, so its expiry reclaims disk by dropping whole segments too — before that it ran a non-segmented backend whose deleteRange expiry reclaimed nothing until a later background compaction happened to cover the tombstone |
| inMemoryWindowStore(name, retention, windowSize, retainDuplicates) | ✅ | In-memory window store. Not segmented: expires per record at exactly retention, and its supplier reports segmentIntervalMs() == 1, as KS's does. |
| persistentSessionStore(name, retentionPeriod) | ✅ | RocksDB session store. Segments are sized max(retention / 3, 60s) — a session has no window size, so KIP-1366's >= windowSize term does not apply — not configurable, and effective retention is up to retention + segmentInterval. On a uniform layout StoatFlow never retains more than Kafka Streams here: its divisor is /3 against KS's /2 and both are floored at 60s, so its interval is never the larger of the two — equal while both sit on the floor, tighter once KS leaves it. (Through a retention narrowing that ordering is suspended, since the older, wider segments below are still live.) Bounds assume the default onWindowUpdate(). See the windowing docs for the regimes and worked examples. A retention change remaps rather than requiring a wipe. After a retention narrowing, existing segments keep their original span, so the store frees disk gradually rather than at once |
| inMemorySessionStore(name, retentionPeriod, expiryIndex) | ✅ | In-memory session store. Not segmented: expires per record at exactly retention, and its supplier reports segmentIntervalMs() == 1, as KS's does. 🆕 The third parameter (default false) opts into an end-time expiry index — no KS equivalent; the two-argument call is unchanged. See expiring sessions from an in-memory store. |
| inMemorySessionStoreWithHeaders(name, retentionPeriod, expiryIndex) | ✅ | In-memory headers-aware session store (KIP-1271). 🆕 Same expiryIndex third parameter as inMemorySessionStore — it builds the same store class. |
| persistentTimestampedKeyValueStore(name) | ✅ | Timestamped KV store (KIP-258) |
| persistentTimestampedWindowStore(name, retention, windowSize, retainDuplicates) | ✅ | Timestamped window store (KIP-258); 4-arg retainDuplicates overload supported. Same segment sizing as persistentWindowStore (segmented whatever the retainDuplicates value, since 2026-09-03) and the same effective-retention bounds. Switching a populated store between this and persistentWindowStore changes the value format and is answered with a full restore |
| inMemoryTimestampedKeyValueStore(name) | ✅ | In-memory timestamped KV store (KIP-258) |
| inMemoryTimestampedWindowStore(name, ...) | ✅ | In-memory timestamped window store (KIP-258). Not segmented: expires per record at exactly retention, and segmentIntervalMs() reports 1, as KS's does — the segmented persistentTimestampedWindowStore keeps up to retention + segmentInterval. |
| persistentVersionedKeyValueStore(name, historyRetention) | ✅ | Versioned store — RocksDB (KIP-889). Point-in-time reads below the retention floor return the latest version if it satisfies the bound, else null — KS-exact. Divergence: the stream time behind the grace window is shared by every lane writing the store, where KS keeps one per task; under exactly_once a write and a read are judged on it as of the lane's own commit. Divergence: retention is a minimum bound. KS drops whole segments; StoatFlow reclaims with a budgeted background sweep (stoatflow.state.versioned.sweep-period-ms), so the on-disk footprint is bounded rather than exact, and a standby applies the active's reclamation tombstones, so its footprint tracks the active within replication lag. Under exactly_once every lane reads it as of its own commit, so any join can read it |
| persistentVersionedKeyValueStore(name, historyRetention, segmentInterval) | ✅ | 3-arg KS overload; segmentInterval accepted but ignored (not segmented) — logs a warning |
WindowBytesStoreSupplier.segmentIntervalMs() / SessionBytesStoreSupplier.segmentIntervalMs() | ✅ | Reports the real interval for the backend the supplier builds — max(retention / 3, 60s, windowSize) for a segmented window store, max(retention / 3, 60s) for a session store, and 1 for an in-memory one, exactly as Kafka Streams' suppliers do |
| inMemoryVersionedKeyValueStore(name, historyRetention) | ✅ | Versioned store — in-memory (KIP-889). 🆕 no KS equivalent. Same read contract and sweep as the persistent backend; its observed stream time is not durable, so it restarts unset. No per-commit write buffer: exactly-once through an engine rebuild of an aborted commit, but its reads are live, so under exactly_once it is refused as a table a join reads from another lane (KSC-126) — the TopologyTestDriver judges the declared store, so it refuses this one and accepts persistentVersionedKeyValueStore |
| lruMap(name, maxCacheSize, tombstoneOnEviction) | 🆕 | StoatFlow extension: with true, each eviction also writes a changelog delete (needs logging; refused with withLoggingDisabled()). The two-argument form keeps Kafka Streams' behaviour (evictions are local only) |
| lruMap(name, maxCacheSize) | ✅ | LRU cache store with bounded entries. No per-commit write buffer (exactly-once through an engine rebuild of an aborted commit); reads are always live — will not honour READ_COMMITTED once KIP-892 isolation-level reads land (KSC-116); refused under exactly_once as a table a join reads from another lane — either side of a primary-key or foreign-key join, a re-keyed stream-table join, a global-table join (KSC-126) |
| keyValueStoreBuilder(supplier, keySerde, valueSerde) | ✅ | Returns a KS StoreBuilder |
| timestampedKeyValueStoreBuilder(supplier, keySerde, valueSerde) | ✅ | KIP-258 |
| windowStoreBuilder(supplier, keySerde, valueSerde) | ✅ | Returns a KS StoreBuilder |
| timestampedWindowStoreBuilder(supplier, keySerde, valueSerde) | ✅ | KIP-258 |
| sessionStoreBuilder(supplier, keySerde, aggSerde) | ✅ | Returns a KS StoreBuilder |
| versionedKeyValueStoreBuilder(supplier, keySerde, valueSerde) | ✅ | KIP-889 |
| timestamped{KeyValue,Window} / sessionStoreWithHeadersBuilder(...) | ✅ | Headers-aware builders (KIP-1271); pass a *WithHeaders supplier |
| KeyValue/Window/Session/VersionedBytesStoreSupplier, StoreBuilder | ✅ | KS supplier/builder interfaces are present |
KeyLockManager (StoatFlow extension)
Striped lock utility for multi-key atomic sections in user-defined Processors. Not in Kafka Streams.
| Method | Status | Notes |
|---|---|---|
| withLock(key, block) | 🚀 | Single-key lock scope |
| withLocks(keys, block) | 🚀 | Multi-key lock scope with deadlock-free ordering |
Record headers in state stores (KIP-1271 / KIP-1285)
Opt-in persistence of a record's Headers alongside key/value, across the timestamped KV, versioned, timestamped-window, and session store families.
| Surface | Status | Notes |
|---|---|---|
| Materialized.withRecordHeaders() | 🆕 | Per-store opt-in (DSL) |
| dsl.store.format = HEADERS | ✅ | Global default to headers-aware stores |
| Stores.*WithHeaders(...) suppliers + *WithHeadersBuilder(...) | ✅ | Processor-API headers stores (all four families) |
| QueryableStoreTypes.*WithHeaders() | ✅ | Read stored headers back via IQ |
| SessionStoreWithHeaders.expireSessions(watermark) | 🚀 | StoatFlow extension (abstract). StoatFlow expires sessions from the aggregation, not inside the store, so a custom implementation must implement it. Kafka Streams' interface has no such method |
| In-place upgrade (existing store → headers-aware) | ✅ | Flip on, restart — no wipe, no changelog replay |
| Downgrade refusal (headers-aware → plain) | ✅ | Like KS, a downgrade is refused with a diagnosed, actionable message rather than a raw RocksDB error — it is a declared format change, not corruption |
state.format.downgrade = wipe-and-restore | 🚀 | StoatFlow-only: acknowledge the downgrade in config and it rebuilds that store from the changelog. KS offers no equivalent — its only remedy is deleting local state by hand |
| Empty-keyspace downgrade (nothing written in headers mode) | 🚀 | Dropped in place, free. KS refuses this case too |
| StreamJoined headers | ❌ (matches KS) | KS does not persist headers on join buffers |
Interactive Queries
StoatFlow (entry point) methods
| Method | Status | Notes |
|---|---|---|
| store(StoreQueryParameters) | ✅ | Returns a read-only view, as in Kafka Streams, in one of two ways. When the store holds the value shape the matcher promises, the view keeps the store's interface and every write throws UnsupportedOperationException — including through sessionStoreWithHeaders() and versionedKeyValueStoreWithHeaders(). When the matcher converts (keyValueStore() over a timestamped store, the headers-aware twins, sessionStore() over a headers-aware session store), the view implements only the ReadOnly* interface and has no write methods: a cast to a writable store type fails with ClassCastException. Either way the handle looks the store up on every read, so it survives an in-place engine restart; QueryableStoreType.create(StateStoreProvider, String) has Kafka Streams' signature |
| storeNames() | ✅ | List available store names |
| queryMetadataForKey(store, key, serializer) | ✅ | Returns KeyQueryMetadata — single-instance: the local host, no standbys, the key's faithful Kafka partition |
| queryMetadataForKey(store, key, headers, serializer) | ✅ | KIP-1321: the key serialized with the headers, as at the sink |
| queryMetadataForKey(store, key, headers, partitioner) | ✅ | KIP-1321: the partition your StreamPartitioner returns; anything but a single partition throws IllegalArgumentException. ⚠️ Optional.empty() also throws (KS returns metadata without a usable partition) |
| streamsMetadataForStore(store) | ✅ | The single local host materializing the store |
| metadataForAllStreamsClients() | ✅ | The one local instance |
| setUncaughtExceptionHandler(handler) | ✅ | KS StreamsUncaughtExceptionHandler; REPLACE_THREAD triggers an in-place engine restart (fault-storm-guarded — repeated faults escalate to shutdown), under hot-standby HA as well as without it: the fault is absorbed in place and the pod keeps the active role; SHUTDOWN_CLIENT/SHUTDOWN_APPLICATION shut the single instance down |
| pause() | 🚀 | Pause processing with drain semantics |
| unpause() | 🚀 | Resume processing after pause |
| awaitState(state, timeout) | 🚀 | Wait for target state with timeout |
| stateTransitionHistory() | 🚀 | Recent state transitions with timestamps |
StoreQueryParameters
| Method | Status | Notes |
|---|---|---|
| fromNameAndType(name, type) | ✅ | Factory method |
| enableStaleStores() | ✅ | Allow queries during RESTORING |
| withPartition(partition) | 🛑 | N/A for single-instance |
QueryableStoreTypes
| Method | Status | Notes |
|---|---|---|
| keyValueStore() | ✅ | Returns ReadOnlyKeyValueStore. Over a timestamped store (the default for grouped and windowed aggregations) or its headers-aware twin it answers with the plain value, as in Kafka Streams |
| windowStore() | ✅ | Returns ReadOnlyWindowStore. Over a timestamped window store or its headers-aware twin it answers with the plain value, as in Kafka Streams |
| sessionStore() | ✅ | Returns ReadOnlySessionStore. Also accepts a headers-aware session store and answers with the plain aggregate, as in Kafka Streams |
| timestampedKeyValueStore() | ✅ | Returns ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>> (KIP-258). Also accepts the headers-aware twin and drops the headers, as in Kafka Streams. Answers every DSL key-value store — table(), toTable(), globalTable(), KTable operators and cogroup aggregates included |
| timestampedWindowStore() | ✅ | Returns ReadOnlyWindowStore<K, ValueAndTimestamp<V>> (KIP-258). Also accepts the headers-aware twin and drops the headers, as in Kafka Streams |
| versionedKeyValueStore() | ✅ | Returns ReadOnlyVersionedKeyValueStore<K, V> (KIP-889) |
The global state model eliminates partition routing — all state is locally accessible from a single process, so
withPartition() and inter-instance RPC are unnecessary.IQ metadata (single-instance: always local)
Kafka Streams' host/partition-distribution metadata types exist so cross-instance "find the host, then query" code compiles and runs. Under StoatFlow's single-instance model the answer is always "this host, locally."
| Type | Status | Notes |
|---|---|---|
| HostInfo(host, port) | ✅ | The local host; sourced from application.server (synthetic localhost:0 if unset) |
| StreamsMetadata | ✅ | hostInfo(), stateStoreNames(), topicPartitions(); no standbys |
| KeyQueryMetadata | ✅ | activeHost() = local, no standby hosts, partition() = the key's faithful Kafka partition |
Exception handlers
| Handler | Status | Notes |
|---|---|---|
| Source deserialisation with headers | ✅ | Source deserializers receive the record headers (deserialize(topic, headers, data)), as in Kafka Streams — header-aware serdes work at the source |
| DeserializationExceptionHandler | ✅ | KS-ordered handle(ErrorHandlerContext, ConsumerRecord, Exception); LogAndFail (default), LogAndContinue, DLQ |
| ProcessingExceptionHandler | ✅ | KS-ordered handle(ErrorHandlerContext, Record, Exception); LogAndFail (default), LogAndContinue, DLQ. A FAIL verdict a processor catches and rethrows wrapped is not offered to the handler again — Kafka Streams re-offers it at the wrapping node |
| ProductionExceptionHandler | ✅ | KS-ordered handle(ErrorHandlerContext, ProducerRecord, Exception); intelligent retry |
| ErrorHandlerContext | ✅ | KS interface: topic()/partition()/offset()/timestamp()/headers()/processorNodeId()/taskId() |
| StreamsException | ✅ | Extends KafkaException; StoatFlow's fatal failures extend it, so catch (StreamsException) works |
Lifecycle
| Feature | Status | Notes |
|---|---|---|
| State enum | ✅ | StoatFlow keeps a richer set than KS's 7 — CREATED, STARTING, VALIDATING_STATE, RESTORING, RUNNING, DRAINING, PAUSED, STOPPING, STOPPED, ERROR. For a KS-shaped KafkaStreams.State, use the KafkaStreamsState mirror + State.toKafkaStreamsState() / kafkaStreamsState() (KSC-81) |
| StateListener | ✅ | KS-exact single setStateListener(StateListener) — a Java method reference / lambda resolves with no cast (KSC-82); an existing BiConsumer<State,State> migrates via StateListener.of(...) |
| StateRestoreListener | ✅ | Full KS-compatible: onRestoreStart / onBatchRestored / onRestoreEnd |
| StreamsUncaughtExceptionHandler | ✅ | setUncaughtExceptionHandler(); REPLACE_THREAD = in-place engine restart (fault-storm-guarded; works under hot-standby HA too — absorbed in place, no failover until the budget is spent), SHUTDOWN_* = single-instance shutdown |
| cleanUp() | ✅ | Deletes all local state for the application id, lock files included; valid only in CREATED/STOPPED; rebuilt from changelog on next start. Unlike Kafka Streams 4.4 no process-metadata file is retained, so a full reset needs no manual delete |
| close() | ✅ | Graceful shutdown; blocks until terminal. Every concurrent close caller now blocks until completion (a losing second close() no longer returns early mid-shutdown) — KSC-83 |
| close(Duration) | ✅ | The timeout bounds only the caller's wait, never the shutdown work (KS-exact). Duration.ZERO = async signal-and-return; negative → IllegalArgumentException; shutdown continues after a false return. Returns true on any terminal state incl. ERROR (divergence — KS only NOT_RUNNING) — KSC-83 |
| close(CloseOptions) | ✅ | Mirrors the top-level KIP-1153 io.stoatflow.core.CloseOptions (KS-exact defaults + mutate-and-return-this fluents + @JvmStatic factories). GroupMembershipOperation has the Kafka Streams 4.4 shape — DEFAULT, LEAVE_GROUP, REMAIN_IN_GROUP, with DEFAULT as the default and what close() passes. An explicit LEAVE_GROUP/REMAIN_IN_GROUP always wins over the leaveGroupOnClose config; DEFAULT follows it (default: remain). DEFAULT never leaves the group unless leaveGroupOnClose is set, because StoatFlow does not use the streams group protocol — KSC-112. No-op under HA assign(). The deprecated nested KIP-812 KafkaStreams.CloseOptions is not mirrored — a port rewrites one line to the factories — KSC-83 |
Streams group protocol (KIP-1071 family): group.protocol=streams, the topology-description push (KIP-1331), Admin.describeStreamsGroups | 🛑 | Not applicable — StoatFlow is a classic consumer group, so describeStreamsGroups never lists it. Pull the topology description from GET /topology/ks instead — KSC-113 |
Configuration
| Aspect | Status | Notes |
|---|---|---|
| StreamsConfig builder | ✅ | Type-safe builder pattern (StreamsConfig.builder(appId, bootstrap).…build()) — the primary config model |
Properties / *_CONFIG constants | ✅ | Canonical KS adapter (ADR-124). StreamsConfig.fromProperties(props) / fromMap(map) adapt a KS-keyed Properties/Map onto the typed builder with complete coverage of the KS-4.3.0 overlap surface — all 74 keys, listed one by one in KS config-key mapping below: typed mappings, class-name pluggables (serdes, the three exception handlers, default.client.supplier, rocksdb.config.setter), client prefixes (consumer./main.consumer./restore.consumer./producer./admin.), unprefixed client keys routed by per-client validity exactly as Kafka Streams does (security.protocol reaches consumer + producer + admin, acks only the producer; a prefixed spelling wins, and StoatFlow's forced overrides win over both — main.consumer. is a main-consumer-only scope, as in KS, not an alias for consumer.), and a lenient 3-tier unknown-key policy (mapped / recognised-no-op / unknown) + opt-in stoatflow.config.strict, which throws on truly-unknown keys instead of forwarding them to every client as custom props. The typed builder/YAML stays the recommended model; the adapter is the canonical KS-compat surface, not a stepping stone to a Map-backed swap (ADR-124). The KS *_CONFIG symbol constants and prefix helpers (consumerPrefix(...) … adminClientPrefix(...)) are also shipped for symbol-level source compatibility (KSC-80). |
StreamsConfig(Map) / StreamsConfig(Properties) ctor | ✅ | KSC-79. The KS-faithful new StreamsConfig(props) / new StreamsConfig(map) bootstrap idiom compiles after the import-swap (Properties binds the single Map<*,*> ctor); it delegates to the same fromMap(...).build() adapter as fromProperties. A missing application.id throws at construction (IllegalArgumentException, where KS throws ConfigException). The typed builder(...) / fromProperties(...) paths stay recommended. |
toProperties() | 🆕 | StreamsConfig.toProperties() emits the config back to KS-keyed Properties (resolved-view, masked) — for /config, diffing, inspect/reconstruct |
default.timestamp.extractor | ✅ | StreamsConfig.defaultTimestampExtractor — a global fallback TimestampExtractor for sources with no per-Consumed WatermarkStrategy (event-time basis only; the watermark default is unaffected) |
| Admin client config | ✅ | StreamsConfig.adminConfig / admin.* prefix — passthrough merged at all AdminClient sites so a secured cluster authenticates |
| KafkaClientSupplier | ✅ | Custom client factory for instrumentation/testing |
| DefaultKafkaClientSupplier | ✅ | Standard KafkaConsumer/KafkaProducer/Admin factory |
| RocksDB config | ✅ | Bounded memory by default |
| Serdes | ✅ | Compatible |
| Default key/value Serdes | ✅ | defaultKeySerde/defaultValueSerde propagate through the DSL |
| Consumer/Producer config | ✅ | Pass-through maps |
| application.server | ✅ | host:port (or protocol://host:port) for the local host reported by interactive-query metadata; defaults from the runtime HTTP host:port when unset. A malformed value fails at startup, as in Kafka Streams 4.4 |
| state.cleanup.dir.max.age.ms (KIP-1259) | ✅ | Age-gated purge of stale local state at startup (rebuilt from changelog); default disabled |
| state.cleanup.delay.ms | ✅ | Orphaned-store cleanup interval |
KS config-key mapping
Full Kafka Streams 4.3.0 StreamsConfig ConfigDef coverage — all 74 keys — as accepted by StreamsConfig.fromProperties / fromMap (ADR-124). A build-time drift test pins this set against the real Kafka Streams ConfigDef, so a key added upstream cannot go unnoticed. Some rows below cover a family of keys.
Status in this table: ✅ mapped · 🆕 added (ADR-124) · 🟡 recognised-no-op (warns, never throws — even under stoatflow.config.strict) · 🧩 client pass-through (routed to every client whose own configNames() contains it, exactly as Kafka Streams does; a consumer. / producer. / admin. prefix still wins, and StoatFlow's forced overrides win over both).
A port that references the symbol rather than the raw string is covered too (KSC-80): StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, the six client-prefix constants (CONSUMER_PREFIX / MAIN_CONSUMER_PREFIX / RESTORE_CONSUMER_PREFIX / GLOBAL_CONSUMER_PREFIX / PRODUCER_PREFIX / ADMIN_CLIENT_PREFIX) and the prefix helpers (StreamsConfig.consumerPrefix(prop) … adminClientPrefix(prop)) all ship, and the keys they build route or no-op exactly as the raw strings do. TOPIC_PREFIX and the topology-optimization value constants are not shipped.
| KS key | Status | StoatFlow mapping / reason |
|---|---|---|
application.id | ✅ | applicationId |
bootstrap.servers | ✅ | bootstrapServers |
state.dir | ✅ | stateDir (StoatFlow default /var/stoatflow/state) |
default.key.serde | ✅ | defaultKeySerde (Class or class-name) |
default.value.serde | ✅ | defaultValueSerde (Class or class-name) |
num.stream.threads | ✅ | → numLanes (closest single-instance analogue) |
processing.guarantee | ✅ | processingGuarantee (plus the exactly_once_beta alias; StoatFlow defaults to EXACTLY_ONCE where KS defaults to at_least_once — KSC-84) |
application.server | ✅ | applicationServer (interactive-query metadata host:port — KSC-47) |
deserialization.exception.handler (+ deprecated default.deserialization.exception.handler) | 🆕 | deserializationExceptionHandler (class-name) |
production.exception.handler (+ deprecated default.production.exception.handler) | 🆕 | productionExceptionHandler (class-name) |
processing.exception.handler | 🆕 | processingExceptionHandler (class-name) |
default.timestamp.extractor | 🆕 | defaultTimestampExtractor (global fallback, event-time basis) |
default.client.supplier | 🆕 | kafkaClientSupplier (class-name) |
rocksdb.config.setter | 🆕 | rocksDbConfigSetter (class-name) |
dsl.store.format | 🆕 | dslStoreFormat (DslStoreFormat.fromConfigString) |
commit.interval.ms | 🆕 | → barrierIntervalMs (a semantic seed; the commit cadence is adaptive) |
buffered.records.per.partition | 🆕 | bufferedRecordsPerPartition — a pause threshold, as in Kafka Streams (any positive value) |
max.task.idle.ms | 🆕 | maxTaskIdleMs (1:1; both allow -1) |
replication.factor | 🆕 | changelogReplicationFactor |
state.cleanup.delay.ms / state.cleanup.dir.max.age.ms | 🆕 | stateCleanupDelayMs / stateCleanupDirMaxAgeMs (KIP-1259) |
ensure.explicit.internal.resource.naming | 🆕 | ensureExplicitNaming (StoatFlow defaults to true, KS to false; the mapping applies only when the key is present) |
processor.wrapper.class | ⚠️ | KIP-1112 — supported. One ProcessorWrapper wraps every node, DSL and Processor API alike, at topology-compile time. Full KS fidelity for Processor API nodes; DSL-internal nodes get the record path only, and replacing an internal node is rejected at build time. A wrapper set on the runtime config alone is honoured, where KS ignores that case as too late |
num.standby.replicas | 🟡 | Single-instance: no standby replicas |
acceptable.recovery.lag / max.warmup.replicas / task.assignor.class / probing.rebalance.interval.ms | 🟡 | Single-instance: no standby, warmup, assignor or probing |
rack.aware.assignment.* / upgrade.from | 🟡 | Single-instance: no rack-aware assignment, no rolling-upgrade protocol |
group.protocol | 🟡 | StoatFlow manages consumer-group membership internally |
topology.description.push.enabled (KS 4.4) | 🟡 | Ignored with a warning: StoatFlow is not a streams group, so there is no push. The same topology description is available as a pull from GET /topology/ks — KSC-113 |
statestore.cache.max.bytes / cache.max.bytes.buffering | 🟡 | StoatFlow's cache caps are per-store and commit-triggering (caching.max-entries / caching.max-estimated-bytes), not a global evict budget; global memory is bounded by state.uncommitted-max-bytes |
client.id | 🟡 | Client ids are derived from application.id; override with consumer.client.id / producer.client.id |
task.timeout.ms | 🟡 | Commit-path waits are bounded by barrierTimeoutMs instead |
topology.optimization | 🟡 | Re-keying is in-memory; KTable source reuse is autoReuseKTableSourceTopics |
repartition.purge.interval.ms | 🟡 | In-memory re-keying: there are no repartition topics to purge |
processing.exception.handler.global.enabled | 🟡 | The processing handler is already applied globally |
errors.dead.letter.queue.topic.name | 🟡 | The DLQ is configured per handler (DeadLetterQueue*ExceptionHandler) |
default.dsl.store / dsl.store.suppliers.class | 🟡 | Choose the store with Materialized.withStoreType |
default.list.{key,value}.serde.{inner,type} | 🟡 | No ListSerde default-type config; use explicit serdes |
windowstore.changelog.additional.retention.ms | 🟡 | Window changelogs use compaction; no extra retention |
window.size.ms / windowed.inner.class.serde | 🟡 | Deprecated in KS; use the typed WindowedSerdes |
built.in.metrics.version / metric.reporters / metrics.* / enable.metrics.push | 🟡 | Metrics go through Micrometer, not the Kafka metrics subsystem |
config.providers | 🟡 | Resolve externalised config before constructing StreamsConfig |
allow.os.group.write.access | 🟡 | StoatFlow does not relax state-dir permissions via config |
log.summary.interval.ms | 🟡 | No periodic summary-log toggle |
poll.ms | 🟡 | Use consumerPollTimeoutMs |
security.protocol / connections.max.idle.ms / metadata.max.age.ms / {receive,send}.buffer.bytes / request.timeout.ms / reconnect.backoff.* / retry.backoff.ms / metadata.recovery.* | 🧩 | Client properties, routed unprefixed to every client they are valid for (KSC-92, KS parity): security.protocol reaches consumer, producer and admin; the rest go by their own ConfigDef. A consumer. / main.consumer. / producer. / admin. prefix overrides the bare form. The same routing applies to bare ssl.* and sasl.*, and to producer-only keys such as acks |
Key architectural differences
The runtime-model deltas — single instance vs. rebalancing cluster, lanes vs. partition-bound tasks, in-memory re-keying vs. repartition topics, barrier-based exactly-once, global vs. partition-scoped state — are summarized in the canonical deltas-at-a-glance table, with the conceptual treatment on the same page. The rows below are the operational and time-semantics differences that surface while reading this matrix:
| Aspect | Kafka Streams | StoatFlow |
|---|---|---|
| RocksDB memory | Unbounded default | Bounded default (256 MiB) |
| Window closure | Stream-time based | Watermark-based |
| Grace period | Accepts a late record while windowEnd + grace is ahead of the task's stream time (max observed timestamp, KIP-328); the same clock closes the window | One clock, the watermark (min across non-idle partitions − out-of-orderness): a window accepts records until it passes windowEnd + grace (a session: last event + gap + grace), and then closes — final emission, suppress release, join null side, event-time timers (KSC-132) |
| Watermark strategy | N/A | Flink-style WatermarkStrategy |
| Punctuators | Task-scoped, on the stream thread | Dedicated punctuation lane per sub-topology, full state access |
| Key-based timers | N/A | Flink-inspired TimerService |
| Scheduled sources | N/A | Interval & cron-based emission |
| Runtime control | N/A | pause()/unpause() |
Test utilities
TopologyTestDriver drives a topology in-memory (no broker) for unit tests. All state stores — including RocksDB-Materialized ones — run in-memory in the test driver, but the exactly-once store checks judge the store you declared (KSC-126: a RocksDB versioned join input passes, an in-memory versioned or lruMap one is refused, as in production); real-engine + RocksDB end-to-end coverage uses integration tests with an embedded broker.
TopologyTestDriver
| Member | Status | Notes |
|---|---|---|
new TopologyTestDriver(topology) / (topology, Properties) / (topology, Instant) / (topology, Properties, Instant) | ✅ | KS-shaped constructors; Properties via StreamsConfig.fromProperties |
createInputTopic(name, Serde/Serializer, …) | ✅ | Serde and Serializer overloads + 5-arg (…, Instant, Duration) start-time/auto-advance form (serdes accepted, ignored) |
createOutputTopic(name, Serde/Deserializer, …) | ✅ | Serde and Deserializer overloads |
getAllStateStores() / producedTopicNames() | ✅ | All stores by name; the set of produced topics |
advanceWallClockTime(Duration) / get*Store(name) / close() | ✅ | advanceWallClockTime also moves currentSystemTimeMs() and TimerService.currentProcessingTime() |
TestRecord
KS positional order (key, value, headers, timestamp); ctors (K,V), (K,V,Headers[,Long]), (K,V,Instant), (K,V,Headers,Instant), value-only (V), (ConsumerRecord), (ProducerRecord), all callable from Java (a null timestamp means unset); bean getters getKey()/getValue()/getHeaders()/getRecordTime() + KS accessors key()/value()/headers()/timestamp(). The timestamp is unset by default, as in Kafka Streams: pipeInput(TestRecord) stamps an unset record with the input topic's current time.
TestInputTopic
| Method | Status | Notes |
|---|---|---|
| pipeInput(key, value , timestamp) | ✅ | timestamp as long or Instant |
| pipeInput(value) / pipeInput(value, Instant) | ✅ | Value-only (null key) — for value-centric / Avro tests |
| pipeKeyValueList / pipeValueList / pipeRecordList | ✅ | Bulk piping; key-value lists use KeyValue; timed (…, Instant, Duration) overloads; pipeRecordList is out-variant |
| advanceTime(Duration) | ✅ | Advance the input topic's record clock |
TestOutputTopic
| Method | Status | Notes |
|---|---|---|
| readKeyValue() / readKeyValuesToList() | ✅ | Return KeyValue<K, V> (.key / .value), not Pair |
| readKeyValuesToMap() | ✅ | |
| readValue() / readRecord() | ✅ |
KeyValue is StoatFlow's KS-shaped key-value type (the same one the DSL uses); a KS port rewrites the KeyValue import to match.
High availability (hot-standby)
Not a KS-parity surface — Kafka Streams scales out with standby tasks; StoatFlow's opt-in hot-standby pairs one active with one or more warm standbys. The surface, and the two deliberate gaps:
| Surface | Status | Notes |
|---|---|---|
stoatflow.ha.mode = off / active-standby | 🚀 | Deploy-time opt-in; off (default) is byte-identical to single-instance |
| Active/passive failover | 🚀 | Crash failover in seconds; graceful deploys / /ha/switch hand off cleanly |
| Split-brain fencing | 🚀 | EOS: broker-enforced producer-epoch fence; ALO: metadata-topic lease |
/ha/{status,switch,promote,demote} endpoints | 🚀 | Operator shims; SWITCH/PROMOTE/DEMOTE return 409 unless a caught-up target exists (?force=true override) |
ha.acceptable-recovery-lag | 🚀 | A single total lag sum (default 50000) — not per-task like KS acceptable.recovery.lag (10000) |
| Continuous changelog replication | 🚀 | The standby streams committed changelog deltas into local state in real time — like KS standby tasks' continuous restore, for the single-active pair |
| Restore-before-process on promotion | 🚀 | A promoting pod restores local state to the committed changelog end before processing |
| Queryable standby (KIP-535 IQ from standby) | ❌ | Standbys are non-queryable; store queries on a standby are rejected |
KS standby tasks / num.standby.replicas | ❌ (different model) | Single-active with warm standbys — not N-instance standby tasks |
See High availability for the operational guide.
Related
- Comparison matrix — feature-by-feature against Kafka Streams and self-hosted / managed Flink
- Building — how-to guidance for the operators above
- KStream and KTable · Aggregations · Windowing · Joins · Processor API
- State stores · Scheduled sources
Maven reference
StoatFlow's Maven support — the stoatflow-bom, the stoatflow-parent convention POM, and the stoatflow-maven-plugin. The three artifacts, every <properties> key, the four goals, and the pinned plugin versions.
Metrics reference
The full catalogue of Prometheus meters StoatFlow exposes on /metrics — name, type, tags, and what each one measures, grouped by area.