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.
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 alpha (
1.0.0-alpha.16); the surface is still moving. 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 |
StreamsBuilder
| Method | Status | Notes |
|---|---|---|
| StreamsBuilder() | ✅ | No-arg |
| StreamsBuilder(TopologyConfig) | ✅ | KSC-77 — seeds build-time defaults (serdes, DSL store format) from a TopologyConfig projection of StreamsConfig. Named-topology / per-topology overrides N/A (single-instance); a non-default processor.wrapper.class throws |
| 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 → sub-topology boundary |
| mapValues(ValueMapper) | ✅ | |
| mapValues(ValueMapperWithKey) | ✅ | Key-aware value mapper variant |
| flatMap(mapper) | ✅ | Key-changing → sub-topology boundary |
| flatMapValues(ValueMapper) | ✅ | |
| flatMapValues(ValueMapperWithKey) | ✅ | Key-aware value mapper variant |
| selectKey(keySelector) | ✅ | Key-changing → sub-topology boundary |
| 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()) |
| processValues(processorSupplier, named?, storeNames...) | ✅ | Non-key-changing; FixedKeyProcessorSupplier + named/declared stores |
| transform() | 🛑 | Removed from KStream in KS 4.x; use process() |
| transformValues() | 🛑 | Removed from KStream in KS 4.x; use processValues() |
KStream — grouping
| Method | Status | Notes |
|---|---|---|
| groupByKey() | ✅ | Returns KGroupedStream |
| groupByKey(Grouped) | ✅ | With Serdes configuration |
| groupBy(keySelector) | ✅ | Key-changing → sub-topology boundary |
| 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; sub-topology boundary |
| 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) |
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 |
| 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() |
| 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 with read-only state on dedicated virtual threads;
PunctuatorMode and the key-based TimerService (which fires in the key's affinity lane 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 read-only state store |
| 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.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 |
| reverseAll() | ✅ | Read-your-writes semantics |
| reverseRange(from, to) | ✅ | Read-your-writes semantics |
| 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) | ✅ | |
| reverseAll() | ✅ | |
| reverseRange(from, to) | ✅ | |
| approximateNumEntries() | ✅ | |
| prefixScan(prefix, serializer) | ✅ | KIP-614 |
WindowStore
Key range queries require K : Comparable<K>. 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 |
| backwardFetch(keyFrom, keyTo, timeFrom, timeTo) | ✅ | Reverse key range; Long and Instant variants |
| 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 |
| backwardFindSessions(keyFrom, keyTo, earliestEndTime, latestStartTime) | ✅ | Key range; Long and Instant variants |
| fetch(key) | ✅ | All sessions for single key |
| backwardFetch(key) | ✅ | Reverse order |
| fetch(keyFrom, keyTo) | ✅ | Key range queries |
| backwardFetch(keyFrom, keyTo) | ✅ | Reverse order |
| 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 |
| 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; every response shuts the instance down (REPLACE_THREAD ≡ SHUTDOWN_APPLICATION) |
| 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(); single-instance: every response shuts the instance down |
| 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 (74 keys): 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.), and a lenient 3-tier unknown-key policy (mapped / recognised-no-op / unknown) + opt-in stoatflow.config.strict that throws only on truly-unknown keys. 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 |
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 MB) |
| 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.