Migration without carrying state

This is the simplest way to move a topology from Kafka Streams to StoatFlow: you do not carry the existing Kafka Streams state across. You point a StoatFlow build at the same source topics under a new application id (a fresh consumer group), let the stateful operators rebuild their state — from the source data or from StoatFlow's own changelog topics, never from the Kafka Streams ones — and switch downstream traffic once the new app has caught up.

It suits most cases: stateless topologies (nothing to carry), topologies whose source topics are still fully retained, and any time you can tolerate a re-read of the input. If your state can only be reconstructed by replaying years of compacted input you can't afford to re-read, see Migration with data migration instead.

When this path applies

The case this path does not cover: state you can only get by replaying source you no longer have (or can't afford to re-read). StoatFlow cannot restore from a Kafka Streams changelog topic — the changelog formats are not interchangeable. That scenario is Migration with data migration.

The four-step cutover

1. Swap the dependency and imports — io.stoatflow:stoatflow-runtime, DSL imports under io.stoatflow.core.topology.*.
2. Swap the entry point — KafkaStreams → StoatFlowRuntime, Properties → application.yaml.
3. Run the StoatFlow app alongside the old one under a fresh application id so it builds an independent consumer group and (for stateful topologies) its own state. Let it catch up to the live offset.
4. Switch downstream traffic to the StoatFlow output, then retire the Kafka Streams app.

Steps 1 and 2 are the code change. Steps 3 and 4 are the operational cutover. The rest of this page walks each one, grounded in the map-filter example that ships in both flavours.

Step 1 — dependency and build

Drop the Kafka Streams dependency and add stoatflow-runtime (the batteries-included module — it transitively pulls in stoatflow-core). If you've followed Installation, the private Maven repository and the io.stoatflow Gradle plugin are already wired up.

Before — the Kafka Streams build:

// build.gradle.kts (Kafka Streams)
plugins {
    kotlin("jvm")
    application
}

application {
    mainClass.set("com.example.MainKt")
}

dependencies {
    implementation("org.apache.kafka:kafka-streams:4.1.1")
    runtimeOnly("ch.qos.logback:logback-classic")
}

After — the StoatFlow build:

// build.gradle.kts (StoatFlow)
plugins {
    kotlin("jvm")
    id("io.stoatflow") version "<stoatflow-version>"
}

stoatflow {
    mainClass.set("com.example.MainKt")
}

dependencies {
    implementation("io.stoatflow:stoatflow-runtime:<stoatflow-version>")
    runtimeOnly("ch.qos.logback:logback-classic")
}

The current version is 1.0.0-alpha.16 — substitute it for the <stoatflow-version> placeholder. The io.stoatflow Gradle plugin applies the JDK 25 toolchain, the --enable-preview and --enable-native-access=ALL-UNNAMED JVM flags, -XX:+UseG1GC, and a runnable fat-jar — so you don't hand-maintain any of the run wiring. See Installation for the Maven equivalent and the plugin-free option.

Step 2 — imports, entry point, and config

The topology code stays the same shape — same StreamsBuilder, same operators, same Consumed / Produced / Named. Two things change: the imports move from org.apache.kafka.streams.* to io.stoatflow.core.topology.*, and the bootstrap moves from a KafkaStreams instance + Properties to StoatFlowRuntime.fromConfig(...) + application.yaml.

Imports

The Kafka serde classes (Serdes.String(), Serdes.Long(), your Avro/Protobuf/JSON serdes) come from the Kafka clients library and carry over verbatim — as does everything else under org.apache.kafka.common.* and org.apache.kafka.clients.* (Headers, ConsumerRecord, ProducerRecord, …). Only the Kafka Streams types move namespaces. See the KS compatibility matrix for the method-by-method DSL parity.

Entry point

In Kafka Streams you build a Properties, construct KafkaStreams, register a shutdown hook, and call start(). The map-filter example does exactly that:

// Before — Kafka Streams
val props = Properties().apply {
    put(StreamsConfig.APPLICATION_ID_CONFIG, "map-filter-example-ks")
    put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, System.getenv("KAFKA_BOOTSTRAP_SERVERS") ?: "localhost:9092")
    put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String()::class.java)
    put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String()::class.java)
}

val builder = StreamsBuilder()
buildTopology(builder)

val streams = KafkaStreams(builder.build(), props)
Runtime.getRuntime().addShutdownHook(Thread { streams.close() })
streams.start()
Thread.currentThread().join()

// Before — Kafka Streams
var props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "map-filter-example-ks");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
    System.getenv().getOrDefault("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092"));
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

var builder = new StreamsBuilder();
buildTopology(builder);

var streams = new KafkaStreams(builder.build(), props);
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
streams.start();
Thread.currentThread().join();

On StoatFlow, StoatFlowRuntime.fromConfig(...) loads application.yaml, starts the HTTP + metrics server, installs graceful shutdown, and runs the topology until terminated. Non-serializable settings — the default serdes, exception handlers — are set in code via streamsConfigOverrides { ... }; everything else lives in YAML:

// After — StoatFlow runtime
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 = { buildTopology(it) },
        configure = {
            streamsConfigOverrides {
                defaultKeySerde(Serdes.String())
                defaultValueSerde(Serdes.String())
            }
        },
    )
    runtime.start()
    runtime.awaitTermination()
}

// After — StoatFlow runtime
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();
        runtime.awaitTermination();
    }
}

The application-id and bootstrap-servers move out of Properties and into application.yaml:

# src/main/resources/application.yaml
stoatflow:
  application-id: map-filter-stoatflow        # NEW id — a fresh consumer group
  bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:-localhost:9092}

  license:
    key: ${STOATFLOW_LICENSE_KEY}

runtime:
  http:
    enabled: true
    port: ${HTTP_PORT:-8080}
  metrics:
    enabled: true

StoatFlow is licensed — stoatflow.license.key is required and has no Kafka Streams equivalent. See License configuration.

Residual manual fixes

The import swap gets a typical port compiling except for a short, enumerated set of deliberate divergences. Walk this checklist once — each is a one-line fix that surfaces at compile time (or, for the last two, at startup):

• JoinWindows.beforeMs / afterMs are getters here, not public fields: replace Java field access jw.beforeMs with jw.getBeforeMs() / jw.getAfterMs(). Kotlin property access is unaffected.
• VersionedRecord.validTo() returns Optional<Long> instead of long — use .orElse(...) / .isPresent().
• Exception handlers receive a Record, not separate key/value: read record.key() / record.value() in ProcessingExceptionHandler.handle(...). Production serialization failures also arrive through handle(...) — branch on (context as ProductionContext).failedOn instead of overriding a separate method.
• StreamPartitioner import is mis-routed by a bulk rewrite: it lives in io.stoatflow.core.topology.StreamPartitioner, not under processor.* as in Kafka Streams — fix that one import by hand. Multicast partitioners (KIP-837) are honoured on sinks but rejected on repartition().
• ValueTransformerWithKey.init takes a FixedKeyProcessorContext instead of ProcessorContext — change the parameter type; state-store access is identical.
• Custom store suppliers must come from Stores.* factories — hand-rolled *BytesStoreSupplier implementations are rejected. Stores.*WithHeadersBuilder(...) needs a *WithHeaders(...) supplier, and an explicit StreamJoined store supplier must set retainDuplicates = true.
• Suppressed maxBytes / withMaxBytes throw (suppress buffers hold objects in memory, not bytes) — use maxRecords(...) or unbounded().
• KStream.transform* is absent (removed in Kafka Streams 4.x too) — use process / processValues.
• A configured processor.wrapper.class throws at startup rather than being silently ignored — remove the key, or apply the wrapper logic directly in your processors.

Two behavioural shifts need no code change but are worth knowing: taskId() is always 0_0 (single instance — don't branch on it), and ProcessorContext.currentStreamTimeMs() returns the watermark — global, and generally lower than KS per-task stream time. The full catalogue, including everything that needs no change, is the KS compatibility matrix.

Step 3 — what to remove, what to keep

The bulk of a Kafka Streams Properties block carries over as stoatflow.kafka.* passthrough or has a direct stoatflow.* equivalent. A handful of properties model a multi-instance cluster and have no meaning in StoatFlow's single-instance model — drop them.

Remove — these model a cluster that doesn't exist

Keep — these carry over

Use a new application id

This is the one config value you must change rather than copy. The application id is StoatFlow's consumer group id (and transactional id). Reusing the Kafka Streams group id would have the StoatFlow app inherit the old committed offsets — it would resume mid-stream and skip the historical records that stateful operators need to rebuild from.

A fresh id gives the StoatFlow app an independent consumer group with no committed offsets, so it can read the source from the beginning. StoatFlow already defaults auto.offset.reset to earliest (matching Kafka Streams), so a stateful topology rebuilds from the start with no extra config — set it explicitly only to make the intent obvious or to choose otherwise:

stoatflow:
  application-id: map-filter-stoatflow        # NOT the old KS application.id
  kafka:
    consumer:
      auto.offset.reset: earliest             # the default; shown here to make the intent explicit

Step 4 — run, catch up, and switch traffic

With the build and config swapped, run the StoatFlow app the same way as any StoatFlow app:

export KAFKA_BOOTSTRAP_SERVERS=localhost:9092
export STOATFLOW_LICENSE_KEY="key/...from your onboarding email..."

./gradlew run

Because the StoatFlow app uses a new application id and writes to its own state and (if stateful) its own changelog topics, it runs independently of the still-running Kafka Streams app — no coordination, no shared group. The recommended cutover:

1. Start the StoatFlow app pointed at the same source topics, with the new id and auto.offset.reset: earliest. It begins consuming from the start of the source and rebuilding state.
2. Write to a parallel output while validating. Either point to(...) at new output topics, or run a side-by-side comparison against the Kafka Streams output. This is the safe default — you confirm correctness before anything downstream depends on it.
3. Watch it catch up. Consumer lag falls toward zero as the StoatFlow app reaches the live tip. Readiness comes up once state restoration completes. What you see: /health/ready returns 200, the stoatflow.consumer.lag.records metric trends to ~0, and /state shows stores fully restored. See Metrics and the REST API.
4. Switch downstream traffic to the StoatFlow output once lag is near zero and the outputs match.
5. Retire the Kafka Streams app. Stop it; once you're confident, delete its consumer group and its internal repartition/changelog topics.

The same cutover as a timeline — the two apps overlap until the switch:

Verify before you cut over

A topology test driver runs the StoatFlow topology in-memory with no broker — use it to confirm the ported topology behaves identically to the Kafka Streams original before any live cutover. The harness mirrors the Kafka Streams TopologyTestDriver:

val driver = TopologyTestDriver.fromBuilder(builder)
val input = driver.createInputTopic<String, String>("input-topic", Serdes.String(), Serdes.String())
val output = driver.createOutputTopic<String, String>("output-topic", Serdes.String(), Serdes.String())

input.pipeInput("k", "streaming")
assertThat(output.readRecord()?.value).isEqualTo("STREAMING")

driver.close()

var driver = TopologyTestDriver.fromBuilder(builder);
var input = driver.createInputTopic("input-topic", Serdes.String(), Serdes.String());
var output = driver.createOutputTopic("output-topic", Serdes.String(), Serdes.String());

input.pipeInput("k", "streaming");
assertThat(output.readRecord().value()).isEqualTo("STREAMING");

driver.close();

Add the test dependency (io.stoatflow:stoatflow-test-utils, test scope — see Installation). Full coverage of the driver, time control, and state-store assertions is in Testing.

Next steps

• Migration with data migration — when you can't re-read the source and must reconstruct state another way.
• How StoatFlow differs from Kafka Streams — the conceptual deltas behind every step here.
• Configuration reference — every stoatflow.* key with type and default.
• KS compatibility matrix — method-by-method DSL parity.
• Production checklist — before the cutover goes live.
• Migrating something non-trivial? Get in touch — real people read every email during the alpha.