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-beta.8). 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 (serdes, DSL store format) and the KIP-1112 ProcessorWrapper from a TopologyConfig projection of StreamsConfig. 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 |
| table(topic, consumed?) | ✅ | |
| table(topic, materialized) | ✅ | Explicit skip-Consumed overload (KS parity) |
| table(topic, consumed?, materialized) | ✅ | With state store |
| globalTable(topic, consumed?) | ✅ | Wraps KTable |
| globalTable(topic, materialized) | ✅ | Without Consumed parameter |
| 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 |
| addReadOnlyStateStore(storeBuilder, topic, consumed, processor) | ✅ | KIP-813: custom processor |
| addReadOnlyStateStore(storeBuilder, topic, consumed) | ✅ | KIP-813: default put/delete semantics |
| 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...) | ✅ | 14 overloads: offset reset, timestamp extractor, deserializers, pattern |
| addProcessor(name, supplier, parents...) | ✅ | User-defined processor with ProcessorSupplier |
| addSink(name, topic, parents...) | ✅ | 8 overloads: serializers, partitioner, TopicNameExtractor |
| addStateStore(builder, processors...) | ✅ | Register store and optionally connect to processors |
| addGlobalStore(...) | ✅ | 2 variants: with/without TimestampExtractor |
| addReadOnlyStateStore(...) | ✅ | 2 variants (KIP-813): source topic is changelog |
| 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() | ✅ | |
| 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 |
| toTable(named?, materialized?) | ✅ | With naming and materialization |
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 |
| 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 |
| 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 |
| 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 |
| 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 |
| 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 |
| join(GlobalKTable, keySelector, ValueJoinerWithKey) | ✅ | Key-aware joiner variant |
| join(GlobalKTable, keySelector, ValueJoiner|ValueJoinerWithKey, Named) | ✅ | Named overloads |
| leftJoin(GlobalKTable, keySelector, ValueJoiner) | ✅ | Stream-GlobalKTable left join |
| leftJoin(GlobalKTable, keySelector, ValueJoinerWithKey) | ✅ | Key-aware joiner variant |
| leftJoin(GlobalKTable, keySelector, ValueJoiner|ValueJoinerWithKey, Named) | ✅ | Named overloads |
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 |
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 |
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 |
SessionWindowedCogroupedKStream
| Method | Status | Notes |
|---|---|---|
| aggregate(initializer, sessionMerger) | ✅ | With Materialized overloads; (initializer, sessionMerger) argument order matches Kafka Streams |
| emitStrategy() | ✅ | OnWindowUpdate (default), OnWindowClose |
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 |
| 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 |
| 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 |
| 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 |
| 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 |
| 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 |
| currentSystemTimeMs() | ✅ | Wall-clock time via System.currentTimeMillis() |
| 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() before any records, close() once at shutdown (KS contract), across all processor paths incl. processValues / transformValues |
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 |
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;
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) |
| 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 |
StreamPartitioner<K, V> | ✅ | KIP-837 partitions(...): Optional<Set<Integer>> (sole method). Sink multicast matches KS; the in-memory repartition path throws on multicast (single-instance). 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; pauses a partition whose watermark drifts beyond the group cap |
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 |
| with(keySerde, valueSerde) | ✅ | |
| keySerde(serde) | ✅ | |
| valueSerde(serde) | ✅ | |
| withCachingEnabled() | ✅ | Suppress downstream emissions (default enabled) |
| withCachingDisabled() | ✅ | Emit all intermediate updates downstream |
| withLoggingEnabled() | ✅ | Enable changelog with optional topic config |
| withLoggingDisabled() | ✅ | Disable changelog for this store |
| withRetention() | ✅ | Override computed retention for windowed stores |
| withStoreType(DslStoreSuppliers) | ✅ | Store type selection |
| withRecordHeaders() | 🆕 | Persist the record's headers alongside key/value in the store (KIP-1271 / KIP-1285) |
| withoutRecordHeaders() | 🆕 | Disable header persistence (default) |
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 |
| withOtherStoreSupplier(storeSupplier) | ✅ | Custom right window store |
| withLoggingEnabled() | ✅ | Enable changelog logging |
| withLoggingEnabled(config) | ✅ | Enable changelog logging with topic config |
| withLoggingDisabled() | ✅ | Disable changelog logging |
| withDslStoreSuppliers(suppliers) | ✅ | Both join stores use same type |
| withStoreName(name) | ✅ | Set base store name |
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 |
| 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 |
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) |
| VersionedRecord<V> | ✅ | Versioned-store record; validTo() returns Optional<Long> |
| KeyValueIterator / WindowStoreIterator | ✅ | Closeable iterators for range/scan queries |
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) | ✅ | |
| 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).
| Method | Status | Notes |
|---|---|---|
| persistentKeyValueStore(name) | ✅ | RocksDB key-value store |
| inMemoryKeyValueStore(name) | ✅ | In-memory key-value store |
| persistentWindowStore(name, retention, windowSize, retainDuplicates) | ✅ | RocksDB window store |
| inMemoryWindowStore(name, retention, windowSize, retainDuplicates) | ✅ | In-memory window store |
| persistentSessionStore(name, retentionPeriod) | ✅ | RocksDB session store |
| inMemorySessionStore(name, retentionPeriod) | ✅ | In-memory session store |
| persistentTimestampedKeyValueStore(name) | ✅ | Timestamped KV store (KIP-258) |
| persistentTimestampedWindowStore(name, retention, windowSize, retainDuplicates) | ✅ | Timestamped window store (KIP-258); 4-arg retainDuplicates overload supported |
| inMemoryTimestampedKeyValueStore(name) | ✅ | In-memory timestamped KV store (KIP-258) |
| inMemoryTimestampedWindowStore(name, ...) | ✅ | In-memory timestamped window store (KIP-258) |
| persistentVersionedKeyValueStore(name, historyRetention) | ✅ | Versioned store — RocksDB (KIP-889) |
| persistentVersionedKeyValueStore(name, historyRetention, segmentInterval) | ✅ | 3-arg KS overload; segmentInterval accepted but ignored (not segmented) — logs a warning |
| inMemoryVersionedKeyValueStore(name, historyRetention) | ✅ | Versioned store — in-memory (KIP-889) |
| lruMap(name, maxCacheSize) | ✅ | LRU cache store with bounded entries |
| 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 |
| 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 read-only store view |
| storeNames() | ✅ | List available store names |
| queryMetadataForKey(store, key, serializer) | ✅ | Returns KeyQueryMetadata — single-instance: the local host, no standbys, the key's faithful Kafka 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 |
| windowStore() | ✅ | Returns ReadOnlyWindowStore |
| sessionStore() | ✅ | Returns ReadOnlySessionStore |
| timestampedKeyValueStore() | ✅ | Returns ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>> (KIP-258) |
| timestampedWindowStore() | ✅ | Returns ReadOnlyWindowStore<K, ValueAndTimestamp<V>> (KIP-258) |
| 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 |
|---|---|---|
| DeserializationExceptionHandler | ✅ | KS-ordered handle(ErrorHandlerContext, ConsumerRecord, Exception); LogAndFail (default), LogAndContinue, DLQ |
| ProcessingExceptionHandler | ✅ | KS-ordered handle(ErrorHandlerContext, Record, Exception); LogAndFail (default), LogAndContinue, DLQ |
| 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; valid only in CREATED/STOPPED; rebuilt from changelog on next start |
| 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 is a per-call override that always wins over the leaveGroupOnClose config (incl. its implicit REMAIN_IN_GROUP default); 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 |
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 for the local host reported by interactive-query metadata; defaults from the runtime HTTP host:port when unset |
| 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 (keep it >= max.poll.records) |
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 |
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 | Retention after close | Extends acceptance window |
| 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; 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() | ✅ |
TestRecord
KS positional order (key, value, headers, timestamp); ctors (K,V), (K,V,Instant), (K,V,Headers[,Instant]), value-only (V), (ConsumerRecord), (ProducerRecord); bean getters getKey()/getValue()/getHeaders()/getRecordTime() + KS accessors key()/value()/headers()/timestamp().
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.