The runtime

stoatflow-runtime is the batteries-included wrapper around the bare stoatflow-core engine. You hand it a topology and a YAML config; it loads the config, wires metrics into the engine, starts an HTTP admin server with health and Prometheus endpoints, bridges your license configuration, runs your plugins, and manages the whole lifecycle — including graceful shutdown on SIGTERM. This is the module most applications depend on.

What the runtime adds over :core

stoatflow-core is the DSL and the processing engine — you can embed it directly with StoatFlow.fromBuilder(config, builder) and drive start() / close() yourself (the KS-shaped close(Duration) / close(CloseOptions) overloads are available too — see the KS compatibility matrix). The runtime wraps that engine and adds the production scaffolding you'd otherwise hand-build:

If you only need the DSL and engine with no scaffolding, depend on stoatflow-core directly. Everything below assumes stoatflow-runtime.

Creating a runtime from YAML

StoatFlowRuntime.fromConfig(...) is the primary entry point. It loads configuration from application.yaml on the classpath and applies it before building your topology. You pass two things:

• topologyBuilder — a function that receives a StreamsBuilder and defines your topology.
• configure — an optional block for everything that can't live in YAML: serdes, exception handlers, and other non-serializable settings, supplied via streamsConfigOverrides { ... }.

import io.stoatflow.core.topology.StreamsBuilder
import io.stoatflow.runtime.StoatFlowRuntime
import org.apache.kafka.common.serialization.Serdes

fun main() {
    val runtime = StoatFlowRuntime.fromConfig(
        topologyBuilder = { builder -> buildTopology(builder) },
        configure = {
            streamsConfigOverrides {
                defaultKeySerde(Serdes.String())
                defaultValueSerde(Serdes.String())
            }
        },
    )

    runtime.start()             // boots engine + HTTP + metrics, then returns
    runtime.awaitTermination()  // block until shutdown
}

private fun buildTopology(builder: StreamsBuilder) {
    builder.stream<String, String>("input")
        .mapValues { it.uppercase() }
        .to("output")
}

import io.stoatflow.core.topology.StreamsBuilder;
import io.stoatflow.runtime.StoatFlowRuntime;
import org.apache.kafka.common.serialization.Serdes;

public class Main {
    public static void main(String[] args) {
        var runtime = StoatFlowRuntime.fromConfig(
            Main::buildTopology,
            builder -> builder.streamsConfigOverrides(cfg -> {
                cfg.defaultKeySerde(Serdes.String());
                cfg.defaultValueSerde(Serdes.String());
            })
        );

        runtime.start();             // boots engine + HTTP + metrics, then returns
        runtime.awaitTermination();  // block until shutdown
    }

    private static void buildTopology(StreamsBuilder builder) {
        builder.<String, String>stream("input")
            .mapValues(v -> v.toUpperCase())
            .to("output");
    }
}

The Java overloads accept Consumer<StreamsBuilder> / Consumer<StoatFlowRuntime.Builder> so void-returning lambdas work without the awkward return Unit.INSTANCE.

Supplying a pre-loaded config

There's a second fromConfig overload that takes an already-loaded ApplicationConfig. Use it when you need to load configuration from a custom source, or to inspect or modify it before the runtime is created:

import io.stoatflow.runtime.config.ConfigLoader

val appConfig = ConfigLoader.loadOrThrow()
// inspect or adjust appConfig here

val runtime = StoatFlowRuntime.fromConfig(
    appConfig = appConfig,
    topologyBuilder = { builder -> buildTopology(builder) },
    configure = {
        streamsConfigOverrides {
            defaultKeySerde(Serdes.String())
            defaultValueSerde(Serdes.String())
        }
    },
)

import io.stoatflow.runtime.config.ConfigLoader;

var appConfig = ConfigLoader.INSTANCE.loadOrThrow();
// inspect or adjust appConfig here

var runtime = StoatFlowRuntime.fromConfig(
    appConfig,
    Main::buildTopology,
    builder -> builder.streamsConfigOverrides(cfg -> {
        cfg.defaultKeySerde(Serdes.String());
        cfg.defaultValueSerde(Serdes.String());
    })
);

Creating a runtime with the builder

fromConfig is the recommended path because it pulls everything from YAML. If you'd rather configure the runtime entirely in code — no application.yaml — use the builder. You're then responsible for constructing the engine's StreamsConfig yourself, and for any HTTP / metrics config you want to override from defaults.

import io.stoatflow.core.config.StreamsConfig
import io.stoatflow.runtime.StoatFlowRuntime
import io.stoatflow.runtime.http.HttpServerConfig

val runtime = StoatFlowRuntime.builder()
    .streamsConfig(
        StreamsConfig(
            applicationId = "my-app",
            bootstrapServers = "localhost:9092",
        ),
    )
    .httpConfig(HttpServerConfig(port = 8080))
    .topology { builder ->
        builder.stream<String, String>("input")
            .mapValues { it.uppercase() }
            .to("output")
    }
    .build()

runtime.start()
runtime.awaitTermination()

import io.stoatflow.core.config.StreamsConfig;
import io.stoatflow.runtime.StoatFlowRuntime;
import io.stoatflow.runtime.http.HttpServerConfig;

var runtime = StoatFlowRuntime.Companion.builder()
    .streamsConfig(
        StreamsConfig.builder("my-app", "localhost:9092").build())
    .httpConfig(new HttpServerConfig(8080, "0.0.0.0", 50, true, true))
    .topology(builder ->
        builder.<String, String>stream("input")
            .mapValues(v -> v.toUpperCase())
            .to("output"))
    .build();

runtime.start();
runtime.awaitTermination();

Metrics aren't configured here — when you omit metricsConfig(...), the builder enables metrics by default and derives the application ID from the StreamsConfig. The builder also exposes metricsConfig(...), addPlugin(...), addLifecycleListener(...), addHealthIndicator(name, indicator), and config setters for infoConfig, configConfig, and healthConfig. The streamsConfigOverrides { ... } block is only used with fromConfig — when you pass a StreamsConfig directly, set everything on that object instead.

Lifecycle

The runtime moves through a small set of states — CREATED → STARTING → RUNNING → STOPPING → STOPPED (or ERROR) — and exposes three lifecycle methods. Query the current state any time with state().

start()

start() boots the runtime and returns (it does not block). Internally it initialises metrics, builds and starts the engine in parallel with HTTP-server setup, starts your plugins, registers the JVM shutdown hook, and transitions to RUNNING. Health endpoints come up early — while the engine is still restoring state — so orchestrator probes get answers during a slow cold start.

start() can only be called from the CREATED state. If engine startup fails, the runtime cleans up everything it had brought up, transitions to ERROR, and rethrows.

awaitTermination()

awaitTermination() blocks the calling thread until the runtime stops. This is what keeps main alive while the engine processes records. If the runtime terminated because of an error, this method rethrows that error — so a non-zero exit on failure falls out naturally:

runtime.start()
runtime.awaitTermination()  // returns on clean stop; throws on error

runtime.start();
runtime.awaitTermination();  // returns on clean stop; throws on error

stop()

stop() shuts the runtime down gracefully: it closes the engine (which drains in-flight records and commits a final barrier), shuts plugins down in reverse registration order, stops the HTTP server, closes the metrics registry, and transitions to STOPPED. You rarely call stop() yourself — the registered JVM shutdown hook calls it on SIGTERM (the signal Kubernetes sends on pod termination), and a watcher thread calls it if the engine terminates on its own. It's idempotent and safe to call from any state.

Lifecycle events

Each transition emits a lifecycle event — PRE_START, POST_START, PRE_STOP, POST_STOP — to any registered RuntimeLifecycleListener. Use these to start or stop dependent services alongside the runtime.

Pause and unpause

Processing can be paused and resumed at runtime through the HTTP admin endpoints — POST /pause and POST /unpause. Pausing stops the runtime consuming new records while keeping the process alive and its endpoints responsive; unpausing resumes from where it left off. This is operated over HTTP rather than as a code method, so you can drive it from an operator, a script, or a runbook without redeploying. See Pause / unpause.

The HTTP admin surface

When runtime.http.enabled is true (the default), start() brings up an HTTP server and registers the operational endpoints. The full catalogue is in the REST API reference; the highlights:

• Health — /health/live and /health/ready for orchestrator probes. See Health checks.
• Metrics — /metrics for Prometheus scraping. See Metrics.
• Introspection — /info, /config (masked), /topology, /topology/ks, /topology/compiled, /state, /watermarks, /offsets, /consumer, and /license.
• Control — /pause, /unpause; plus the /ha/* status and role-switch endpoints when hot standby is enabled.
• Debug — /debug/threads and /debug/barriers for live diagnostics when a topology misbehaves; gated by runtime.http.debug.enabled.

start() returns a handle on the same runtime, so you can chain into it. A couple of accessors are handy from code: httpPort() returns the bound port (useful when you let the server pick one), and meterRegistry() returns the Micrometer registry for custom instrumentation.

Extending the runtime

The runtime is designed to be extended without modifying it:

• Plugins — implement RuntimePlugin to register custom HTTP handlers, metrics, or health indicators during startup, and clean them up on shutdown. Register with addPlugin(...) on the builder, or have fromConfig's configure block add them.
• Lifecycle listeners — implement RuntimeLifecycleListener (a functional interface) to hook the four lifecycle events.
• Custom health indicators — implement HealthIndicator and register via addHealthIndicator(name, indicator) or PluginContext.healthIndicatorRegistry().

The plugin contract, the PluginContext it receives, and worked examples are in Plugins.

Where to go next

• Runtime config — every stoatflow.* and runtime.* YAML key.
• REST API — the full admin-endpoint catalogue.
• Health checks — liveness vs readiness, and the built-in indicators.
• Metrics — the Prometheus surface and what to scrape.
• Plugins — extend the runtime with custom endpoints, metrics, and lifecycle hooks.
• Docker and Native image — package the runtime for deployment.