Engine configuration (:core)

The :core engine knobs every StoatFlow app sets — application id, brokers, default serdes, lanes, commit-barrier cadence, state dir, changelog, processing guarantee, and event-time coordination — with defaults from source.

These are the :core engine settings — the stream-processing knobs that exist whether you embed stoatflow-core directly or run on the batteries-included stoatflow-runtime. On the runtime they live under the stoatflow: block of application.yaml; in plain :core code you set the same values on StreamsConfig (or its Builder). This page groups the ones you'll actually reach for, with their defaults.

This page covers the settings most apps touch. Many more :core knobs exist (dispatch tuning, restoration sizing, RocksDB presets, FK-join, caching, hot-standby HA). For every key with its type and bounds, see Configuration reference. For the layered merge model behind YAML overrides and Kafka client passthrough, see Configuration model and Kafka client config.

Identity and connection

Only application-id is required — it has no default, so the engine won't start without it. bootstrap-servers defaults to localhost:9092, which is fine for local development but you'll set it explicitly for any real broker.

YAML key	`StreamsConfig` field	Default	Notes
`stoatflow.application-id`	`applicationId`	— (required)	Used for the consumer group and the transactional id. Pick a stable value; changing it starts a fresh consumer group.
`stoatflow.bootstrap-servers`	`bootstrapServers`	`localhost:9092`	Comma-separated broker list. Override for any non-local broker.

stoatflow:
  application-id: order-processor
  bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092

StoatFlow is single-instance — exactly one process owns all partitions. Startup validates complete partition assignment and (optionally) checks for other live group members. See Lanes and parallelism and the validation.* block in the reference.

Default serdes

When an operator doesn't specify a serde via Consumed, Produced, Materialized, etc., these defaults apply. Both default to ByteArray — set them to the type your topology mostly uses.

YAML key	`StreamsConfig` field	Default
`stoatflow.default-key-serde`	`defaultKeySerde`	`Serdes.ByteArray()`
`stoatflow.default-value-serde`	`defaultValueSerde`	`Serdes.ByteArray()`

In YAML the values are fully qualified serde class names; in code you pass Serde instances:

val config = StreamsConfig.builder("order-processor", "localhost:9092")
    .defaultKeySerde(Serdes.String())
    .defaultValueSerde(Serdes.String())
    .build()

var config = StreamsConfig.builder("order-processor", "localhost:9092")
    .defaultKeySerde(Serdes.String())
    .defaultValueSerde(Serdes.String())
    .build();

stoatflow:
  default-key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde
  default-value-serde: org.apache.kafka.common.serialization.Serdes$StringSerde

On the runtime, the cleaner way to set typed default serdes is streamsConfigOverrides { ... } in code — see Your first app. Using Schema Registry serdes? Set stoatflow.schema-registry-url; it's propagated to default serdes automatically. More in Serdes.

Setting default-key-serde explicitly does one extra thing: it turns off last-resort boundary key-serde inheritance. StoatFlow cannot distinguish a deliberate Serdes.ByteArray() from an unconfigured default — they are the same value — so "did you configure one at all" is the signal it uses. If you genuinely key a sub-topology boundary with raw bytes, say so here and nothing will second-guess it.

Topology

How the compiler shapes and checks your topology. All of these are StoatFlow-only — Kafka Streams has no equivalent.

YAML key	`StreamsConfig` field	Default
`stoatflow.topology.sub-topology-split`	`subTopologySplit`	`lazy`
`stoatflow.topology.processor-api-key-affinity`	`processorApiKeyAffinity`	`off`
`stoatflow.topology.validation.<rule-id>`	`validationSeverity(rule)`	per rule

sub-topology-split selects where sub-topology boundaries are opened. lazy (the default) opens one only where a downstream operator genuinely needs the re-keyed record's lane affinity — a grouped aggregation, any join, toTable() — plus every explicit repartition(). That is the Kafka Streams repartitionRequired model, so describe() lines up with a KS original. eager restores the pre-1.0.0 boundary at every key-changing node.

Changing it renumbers sub-topology ids, which show up in thread names, the sub_topology metric tag and the topology endpoints. See Lanes and parallelism — in particular the skew note, if you relied on a re-key to spread work across lanes.

processor-api-key-affinity selects whether a Processor API node's key affinity is presumed from the adapter's shape. off (the default) matches Kafka Streams, which never repartitions before process() / processValues() even with connected stores. presumed restores the pre-1.0.0 boundary. It is inert under sub-topology-split: eager, which splits at every key-changing node regardless.

off extends the lazy split's skew trade to Processor API nodes, and after a many-to-one re-key it gives up key affinity there — a real trade-off, spelled out in Key affinity after a re-key. StoatFlow refuses to compile the provable case (papi-key-affinity-presumed, below). It also puts the node in its upstream sub-topology, so it inherits that sub-topology's lane count rather than starting its own — re-check any per-sub-topology numberOfLanes override you set deliberately.

validation.<rule-id> turns individual topology checks off, or promotes them to build failures:

stoatflow:
  topology:
    sub-topology-split: lazy
    processor-api-key-affinity: off
    validation:
      unresolved-boundary-key-serde: error   # off | warn | error
      inherited-boundary-key-serde: off
      papi-key-affinity-presumed: error      # the default for this one

Rule	Default	Fires when
`unresolved-boundary-key-serde`	`warn`	A sub-topology boundary resolved no key serde for the key crossing it and you configured no default key serde, so the implicit `ByteArray` default is used for lane hashing. A boundary straight behind an `addSource(name, keyDeserializer, …)` source counts as unresolved too — declare the source with `addSource(name, Consumed.with(…), …)`. With a default configured, there is nothing to report and the rule stays quiet at every severity
`inherited-boundary-key-serde`	`warn`	The boundary resolved only by inheriting a serde from before the re-key
`papi-key-affinity-presumed`	`error`	A proven re-key (`selectKey`, `map`, `groupBy`, …) feeds a Processor API node connecting a writable store, with no boundary between them. Read-only stores (KIP-813) are excluded — they cannot be written, so there is no update to lose
`papi-key-affinity-presumed-chain`	`warn`	The same shape, but the feeder is itself a Processor API node. `process()` / `addProcessor()` declare a key change unconditionally, so the re-key may not exist at all — refusing the build would refuse topologies Kafka Streams runs happily
`timer-key-lane-affinity`	`warn`	A Processor API node registers timers whose key bytes may differ from the bytes its records are laned by — behind a source that declares only a key deserializer, or entered across a boundary laned by another key serde — so a key's timers may run on another lane than its records. Declare the key serde with `addSource(name, Consumed.with(…), …)`. At runtime a timer for the record's own key that picks another lane follows `timer.lane-affinity.guard`, `error` by default (Timer key serde)
`alignment-idleness-off`	`error`	A source with watermark alignment has no idleness timeout: the global `idleness-timeout-ms` is `0` and the source sets no `withIdleness(...)`, a custom strategy's `idlenessTimeout()` returns zero, or the source uses `withIdleness(ChronoUnit.FOREVER.duration)`. A lagging source that never sends again would then hold every source for good. Add `withIdleness(...)` to the group's sources — a per-source timeout overrides the global `0`, so idleness can stay off elsewhere
`alignment-mixed-watermark-bounds`	`warn`	Sources of one alignment group whose out-of-orderness bounds differ by more than a source's drift (`forMonotonousTimestamps()` counts as 1 ms; custom strategies are not judged). The smaller-bound source stays held, and the application waits whenever the larger-bound source is momentarily empty. Give the group one bound, or raise the drift above the difference
`declared-serde-over-default`	`off`	A DSL operator writes bytes with a serde declared upstream where Kafka Streams would use the configured default, and the two differ: `groupByKey(Grouped)` without serdes, a stream-stream join without `StreamJoined` serdes (its stores and its output key), a cogroup without a `Materialized` key serde, `merge`, a `KTable` `filter`/`mapValues`/`transformValues` store with a `Materialized` that sets no serde. Each is reported where the bytes are written — a sink without `Produced`, a table, an aggregation store; a windowed key always differs (KAFKA-9259). Set it to `error` before a state-carrying migration from Kafka Streams — each site would write other bytes than Kafka Streams did (Serdes)
`presumed-cross-lane-read`	`warn`	Under exactly-once, an `lruMap` or in-memory versioned store that a Processor API node may read across lanes — from a punctuator, or a node downstream of one in the same sub-topology, while records write it, or from another sub-topology than its writer; a global or read-only such store is presumed read by every lane. Such a store cannot be read as of the reading lane's commit, so the read could see another lane's uncommitted update. The same read by a DSL operator is refused (Reading a store from another lane)
`presumed-cross-sub-topology-write`	`warn`	Processor API nodes connect one store from more than one sub-topology — a re-key between them splits them — so two writers on lanes that cross commits at different times can lose an update. Kafka Streams forces such processors into one subtopology. A node that only reads the store loses nothing; the same shape among DSL writers is refused (Sharing a store between processors)

The two boundary-serde rules, the chain rule, the timer rule and alignment-mixed-watermark-bounds default to warn, so upgrading changes nothing about what you are told — only where. papi-key-affinity-presumed defaults to error, because the shape it names is silent data loss at runtime rather than a diagnostic, and so does alignment-idleness-off, because the shape it names stops the application for good. All severities are decided at topology compilation rather than at engine start, so TopologyTestDriver reaches them with no broker: at error the driver refuses to build, naming every offending site in one message rather than failing on the first; at warn the same line lands in your own test output. off silences the condition outright. An unknown rule id is rejected at startup rather than silently ignored.

Lanes (parallelism)

Lanes are the engine's virtual partitions — records with the same key always route to the same lane, preserving per-key order while running many keys in parallel. See Lanes and parallelism for the why.

YAML key	`StreamsConfig` field	Default	Notes
`stoatflow.lanes.count`	`numLanes`	`max(2, CPU cores)`	More lanes = more per-key parallelism, at the cost of more queues. Start at the default; I/O-bound workloads commonly go 4×–8× cores — see Tuning.
`stoatflow.lanes.queue-capacity`	`laneQueueCapacity`	`300`	Per-lane queue depth (backpressure). Minimum 32.
`stoatflow.lanes.thread-type`	`laneThreadType`	`VIRTUAL`	`VIRTUAL` (Project Loom) or `PLATFORM` (OS threads). Keep `VIRTUAL` unless you've switched the RocksDB backend to `JNI` and pinning is a concern.

stoatflow:
  lanes:
    count: 32
    queue-capacity: 500

Commit-barrier cadence

The commit barrier is what makes StoatFlow exactly-once: when a barrier flows through the topology, state flushes and the Kafka transaction commits as one atomic unit. The cadence self-tunes at runtime within the bounds you set — interval-ms is the seed, and the engine adapts between min-interval-ms and max-interval-ms. You set the envelope, not the algorithm; how the engine schedules within it is an implementation concern (see Architecture).

YAML key	`StreamsConfig` field	Default	Notes
`stoatflow.commit-barrier.interval-ms`	`barrierIntervalMs`	`500`	Seed interval. The runtime adjusts from here.
`stoatflow.commit-barrier.min-interval-ms`	`barrierMinIntervalMs`	`150`	Floor on barrier-to-barrier time.
`stoatflow.commit-barrier.max-interval-ms`	`barrierMaxIntervalMs`	`5000`	Ceiling on barrier-to-barrier time.
`stoatflow.commit-barrier.interval-factor`	`barrierIntervalFactor`	`1.5`	Multiplier applied to recent commit duration when spacing barriers, clamped to `[min, max]` (`>= 1.0`). Rarely changed.
`stoatflow.commit-barrier.timeout-ms`	`barrierTimeoutMs`	`60000`	Bounds the commit path. Must be greater than `interval-ms`. Also cascades to the Kafka producer's `transaction.timeout.ms`, `max.block.ms`, and `delivery.timeout.ms`.
`stoatflow.commit-barrier.max-epoch-records`	`barrierMaxEpochRecords`	unset (no hard cap)	Optional hard cap on records committed per barrier. Leave unset unless you need a strict bound.

stoatflow:
  commit-barrier:
    interval-ms: 500
    min-interval-ms: 150
    max-interval-ms: 5000
    timeout-ms: 60000

Tighter latency vs. lower overhead. Lowering min-interval-ms and max-interval-ms commits more often (lower end-to-end latency, more transaction overhead). Raising them batches more work per commit (higher throughput, higher tail latency). The defaults are a balanced starting point — tune from observed commit duration. See Tuning.

A stalled commit is not silent: if a barrier can't complete within timeout-ms, the commit aborts and the process restarts cleanly. On restart, state recovers to the last committed barrier — no loss, no duplication under exactly-once. The barrier scheduling cadence and the commit protocol's internals are engine concerns; see Architecture for the boundary, and Exactly-once for the guarantee.

There are several finer self-tuning knobs under commit-barrier.* (e.g. commit-duration-ema-alpha, initial-max-epoch-records). They have sensible defaults and rarely need changing — the full list is in the reference.

Processing guarantee

Controls the commit protocol. Both modes use barriers for state-flush and offset-commit synchronization; the difference is the transaction boundary.

YAML key	`StreamsConfig` field	Default
`stoatflow.processing-guarantee`	`processingGuarantee`	`EXACTLY_ONCE`

Value	Behaviour
`EXACTLY_ONCE` (default)	Kafka transactions commit output, changelog, and consumer offsets atomically. No duplicates or loss on crash recovery. Higher per-commit cost.
`AT_LEAST_ONCE`	Non-transactional producer with `consumer.commitSync()`. Lower commit latency. Output records may be duplicated on crash recovery; windowed aggregations may over-count.

stoatflow:
  processing-guarantee: EXACTLY_ONCE

See Exactly-once for the full semantics and the at-least-once recovery contract.

State and changelog

State stores are RocksDB-backed by default and durably backed by Kafka changelog topics, so state survives restarts. See State stores and State and thread safety.

State directory and memory

YAML key	`StreamsConfig` field	Default	Notes
`stoatflow.state.dir`	`stateDir`	`${java.io.tmpdir}/stoatflow`	Put this on fast local storage (SSD). In Kubernetes, mount a PV — the default temp dir is wiped on pod restart, forcing full restoration.
`stoatflow.state.uncommitted-max-bytes`	`stateStoreUncommittedMaxBytes`	`268435456` (256 MiB)	Global cap on buffered uncommitted state; exceeding it triggers an early barrier.
`stoatflow.state.segment-checkpoint-records`	`segmentCheckpointRecords`	`100000`	Periodic checkpoint cadence (record count) for segmented window/session stores, which run WAL-off and would otherwise be durable only on `close()`. Bounds how much changelog an ungraceful restart — or a hot-standby promotion — re-applies. Whichever-first with `segment-checkpoint-ms`.
`stoatflow.state.segment-checkpoint-ms`	`segmentCheckpointMs`	`30000` (30 s)	The same checkpoint cadence by wall-clock time; whichever-first with `segment-checkpoint-records`.
`stoatflow.rocks-db.preset`	`rocksDbConfig`	`DEFAULT` (256 MiB)	`DEFAULT` / `LOW_MEMORY` (64 MiB) / `HIGH_PERFORMANCE` (1 GiB) total RocksDB memory.
`stoatflow.rocks-db.backend`	`rocksDbBackendType`	`AUTO`	`AUTO` picks `FFM` (Foreign Function & Memory — no virtual-thread pinning) on the JVM and `JNI` under a native image; set `FFM` or `JNI` to force one.
`stoatflow.state.format-downgrade`	`stateFormatDowngrade`	`refuse`	What to do when a store's on-disk format is newer than the topology asks for — today, record headers turned off over a store that carries them. `refuse` fails startup with an actionable message and touches nothing; `wipe-and-restore` acknowledges the downgrade and rebuilds that store from its changelog.
`stoatflow.state.layout-mismatch`	`stateLayoutMismatch`	`refuse`	What to do when a windowed or session store's on-disk layout no longer describes what the topology asks for and the store has no changelog to rebuild from. A store that can be rebuilt always is, without configuration. `refuse` fails startup naming every affected store and the field that moved; `wipe-and-restore` accepts the (unrecoverable) loss and starts those stores empty. See Changing a store's retention.
`stoatflow.state.versioned.sweep-period-ms`	`versionedSweepPeriodMs`	`900000` (15 min)	Target time for one full reclamation pass over a versioned store. Its retention is a minimum bound: an out-of-window point-in-time read stops answering immediately, but the disk space is reclaimed by a rolling sweep running a slice per commit. This is that SLA — watch the result with `stoatflow.store.versioned.sweep.lag.ms`.
`stoatflow.state.versioned.max-scan-bytes-per-barrier`	`versionedSweepMaxScanBytesPerBarrier`	`8388608` (8 MiB)	Primary budget: bytes the sweep may read per commit, per sub-topology, shared across all versioned stores on that sub-topology. Raise it if `stoatflow.store.versioned.sweep.budget.exhausted.total` climbs steadily. Minimum `65536` — the sweep floors each store's slice there, so a smaller lane budget is rejected at startup rather than silently disabling reclamation.
`stoatflow.state.versioned.max-scan-entries-per-barrier`	`versionedSweepMaxScanEntriesPerBarrier`	`50000`	Secondary cap, for stores whose values are small enough that per-entry CPU dominates instead of bytes. Shared across the sub-topology's stores like the byte budget.
`stoatflow.state.versioned.max-deletions-per-barrier`	`versionedSweepMaxDeletionsPerBarrier`	`10000`	Caps reclamation deletes added to one transaction. They count toward `state.uncommitted-max-bytes`, so raising this substantially can start triggering memory-pressure commits.

stoatflow:
  state:
    dir: /var/lib/stoatflow/state
    uncommitted-max-bytes: 268435456
  rocks-db:
    preset: DEFAULT

Changelog topics

YAML key	`StreamsConfig` field	Default	Notes
`stoatflow.changelog.enabled`	`changelogEnabled`	`true`	When on, state mutations are written to changelog topics inside the same transaction.
`stoatflow.changelog.create-topics-if-not-exist`	`createChangelogTopicsIfNotExist`	`true`	Auto-creates compacted changelog topics on startup.
`stoatflow.changelog.replication-factor`	`changelogReplicationFactor`	`-1` (broker default)	Set explicitly (e.g. `3`) for production durability.
`stoatflow.changelog.num-partitions`	`changelogNumPartitions`	`3`	Partition count for auto-created changelog topics. Sets a changelog's write-throughput ceiling and how the store's write load shards across brokers; one partition funnels every write for a store through a single leader. Every partition is also carried by every commit touching that store, so prefer small counts and raise a specific store only on a measurement — with `Materialized.withChangelogNumPartitions(n)`, which also lets you pin a small store back to `1`. Fixed when the topic is created; changing it later is a no-op plus a startup warning.
`stoatflow.changelog.timer-num-partitions`	`timerChangelogNumPartitions`	`-1` (inherit)	Partition count for the timer store's changelog. Timers are not materialized through the DSL, so there is no `Materialized` to hang a per-store override on. `-1` inherits `num-partitions`; any positive value overrides it for the timer store alone.
`stoatflow.changelog.reconcile-topic-configs`	`reconcileChangelogTopicConfigs`	`true`	Bring an existing changelog topic's `cleanup.policy` and `retention.ms` back in line with what its store asks for, on every start. Per-store settings used to be applied only at topic creation, so a changelog created by an older StoatFlow, by the migration tool, or under a different retention kept its original settings forever. Only those two keys are touched, only for window and session stores (a plain key-value or versioned store's changelog is left on `compact` — its latest value must survive regardless of age), and `retention.ms` is only ever widened: a larger value you set yourself is left alone, because narrowing a `compact,delete` topic makes the broker drop segments. Runs whether or not `create-topics-if-not-exist` is set. If the broker denies `alterConfigs`, StoatFlow logs a warning and starts normally.
`stoatflow.state.restoration-enabled`	`stateRestorationEnabled`	`true`	Restore state from changelog on startup when local state is missing.

stoatflow:
  changelog:
    enabled: true
    replication-factor: 3

For production, set changelog.replication-factor to at least 3 and put state.dir on a persistent volume — together they keep restoration fast (delta, not full) across pod restarts. See Kubernetes and the Production checklist.

Which changelog topics an application creates

Every state store in the topology — every store describe() lists — gets <application.id>-<store>-changelog. The engine adds changelogs for stores of its own, which describe() does not list:

Topic	Exists when	Partitions
`<application.id>-stoatflow-watermark-changelog`	The topology has a windowed aggregation, a time-based `suppress` or a stream-stream join. It holds the committed watermark that late records are judged against, so the judgement survives a restart.	1
`<application.id>-<store>-emitfrontier-changelog`	One per `onWindowClose()` aggregation `<store>`: the last window it emitted, so a restart neither re-emits nor skips one.	As its aggregation store
`<application.id>-timer-store-changelog`	The topology registers Processor API timers and the timer backend keeps a changelog (the default).	`timer-num-partitions`

All are compacted. With create-topics-if-not-exist: false, pre-create them along with the store changelogs and grant the application's principal access to them — startup fails and names any that is missing. With changelog.enabled: false the store and watermark changelogs are not used; the timer changelog follows the timer backend's own setting.

Event-time and watermarks

Watermarks track event-time progress for windowing, late-record handling, and event-time timers. The engine auto-detects whether your topology needs event-time tracking and bypasses it on the hot path when it doesn't. See Event time and watermarks.

YAML key	`StreamsConfig` field	Default	Notes
`stoatflow.watermark.max-out-of-orderness-ms`	`maxOutOfOrderness`	`10000` (10 s)	How far out of order events may arrive before being treated as late.
`stoatflow.watermark.idleness-timeout-ms`	`watermarkIdlenessTimeout`	`300000` (5 min)	Partitions idle this long are excluded from the global watermark. The default for sources that do not set their own `WatermarkStrategy.withIdleness(...)`, which overrides it per source. `0` disables idleness detection for those sources — a partition that never receives a record then pins the global watermark forever, no window or session store ever expires, and segmented stores accumulate RocksDB instances without bound (`stoatflow.store.segments`); the engine logs one WARN at start-up.
`stoatflow.watermark.auto-interval-ms`	`autoWatermarkIntervalMs`	`200`	Periodic watermark emission interval.
`stoatflow.processing.timestamp-coordination-mode`	`timestampCoordinationMode`	`AUTO`	`AUTO` (derive from topology), `ENABLED` (force on), `DISABLED` (force off, benchmarking — refused at startup for a topology with windowed aggregations, `suppress` or stream-stream joins).
`stoatflow.processing.max-task-idle-ms`	`maxTaskIdleMs`	`0`	Cross-partition timestamp-ordering wait. `-1` disables (max throughput); `0` waits only while the broker has data; `>0` adds extra wait. Mirrors KS `max.task.idle.ms`.

stoatflow:
  watermark:
    max-out-of-orderness-ms: 10000
    idleness-timeout-ms: 300000
  processing:
    timestamp-coordination-mode: AUTO

timestamp-coordination-mode: AUTO is correct for almost all topologies — override it only for debugging or benchmarking. Per-source watermark behaviour is configured on Consumed with a WatermarkStrategy; these settings are the engine-wide defaults.

Putting it together

A representative production stoatflow: block:

stoatflow:
  application-id: order-processor
  bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092

  default-key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde
  default-value-serde: org.apache.kafka.common.serialization.Serdes$StringSerde

  processing-guarantee: EXACTLY_ONCE

  lanes:
    count: 32
    queue-capacity: 500

  commit-barrier:
    interval-ms: 500
    min-interval-ms: 150
    max-interval-ms: 5000
    timeout-ms: 60000

  state:
    dir: /var/lib/stoatflow/state
    uncommitted-max-bytes: 268435456

  changelog:
    enabled: true
    replication-factor: 3

  watermark:
    max-out-of-orderness-ms: 10000
    idleness-timeout-ms: 300000

Next steps

Configuration reference — every :core key with type, default, and bounds.
Runtime config () — the runtime.* block (HTTP, metrics, endpoints, health).
Kafka client config — passthrough consumer/producer properties and the layered merge order.
Defaults and presets — the opinionated defaults and how to override them.
Tuning — picking lane count, barrier cadence, and memory caps from observed behaviour.

How configuration works

How the StoatFlow runtime loads configuration — application.yaml on the classpath, overlay files, environment variables, system properties, and the layered precedence order.

Runtime configuration (:runtime)

The runtime.* section of application.yaml — HTTP server, metrics, info/config endpoints, health checks — plus logging levels and Kafka client passthrough.