Tutorial 5: Going to production

You’ve written a topology, given it state, and joined two streams. Going from that to a production deploy is mostly about operational concerns: things that don’t show up in the test driver but bite you in production if you ignore them.

Why production is different from the test driver

The test driver gives you a controlled environment where everything runs in a single process, state lives in memory and disappears when the process exits, there are no network partitions or broker unavailability or slow consumers, and exactly one thread processes everything so there are no race conditions.

Production is the opposite. Your topology will run across multiple instances, each with multiple threads. Network issues happen. Brokers restart. Consumers fall behind. State must survive restarts. The checklist below addresses the gaps between “it works in the test driver” and “it runs reliably in production.”

The eight things you need

Each item below prevents a specific class of production incident:

1. Pick a processing guarantee: at-least-once or exactly-once.
2. Name every stateful operator explicitly.
3. Keep restores short: persistent stores, stable state directories.
4. Wire listeners and progress to your observability stack.
5. Add the orphan-topic detector at startup.
6. Capture a golden file of the topology shape in CI.
7. Test rebalance and restart before they happen for real.
8. Decide whether you need Riffle extensions.

The rest of this page expands each one.

1. Processing guarantee

In the test driver, every record is processed exactly once because there’s only one thread and no failures. In production, instances restart, network partitions happen, and the consumer group rebalances. Without an explicit choice, you might get duplicate records (at-least-once) when a task fails over, or pay overhead you don’t need (exactly-once) for idempotent workloads.

StreamsConfig.processingGuarantee picks between:

	At-least-once (default)	Exactly-once-V2
Output records	Possibly duplicated on rewind	Atomic with offset commit
Side effects (peek, foreach, mapValuesM)	Replay on rewind	Replay on rewind (still!)
Throughput	Higher	Slightly lower (transactional producer overhead)
When to pick it	Your downstream is idempotent	You need exactly-once into Kafka, and you handle side effects separately

At-least-once is fine for most things. Pick it if downstream consumers are already idempotent or if you can tolerate occasional duplicates.

Pick exactly-once when you write to Kafka and a duplicate would matter (financial postings, invoice generation, downstream analytics that can’t dedupe).

For external-system writes that need to be atomic with the Kafka side, see Exactly-once across Kafka and other systems. That’s where the Riffle two-phase-commit sink interface comes in.

2. Name every stateful operator

Auto-generated names change when you add or remove operators in your topology. When a name changes, the runtime creates a new internal topic (changelog or repartition) and abandons the old one. Your previous state is orphaned, queries return empty results, and you have zombie topics consuming disk forever.

The library generates names for unnamed operators automatically. Those names become part of the internal topic layout: <applicationId>-<storeName>-changelog, <applicationId>-<nodeName>-repartition.

If you reshuffle your topology (insert a new map upstream of an existing aggregation, for example), the auto-generated names can shift. The broker creates new internal topics and abandons the old ones. The old ones don’t go away; they sit on the broker holding your previous state.

Every operator that owns a state store should pass an explicit Named or a custom materializedAs (storeName "..."). Anything that creates a repartition topic (repartition, through) should also have a stable name.

-- Good: pinned names
F.count (Mat.materializedAs (storeName "view-counts"))

-- Risky: auto-named, will shift if topology changes
F.count Mat.defaultMaterialized

Full discussion in Topology evolution and rolling deploys.

3. Keep restores short

When an instance restarts (deploy, failure, or scale-up), the new owner of each task restores its stores before it processes anything. Without a standby or a local checkpoint that is the entire changelog: for a 1 TB store at 100 MB/s, three hours during which that task’s partitions don’t move and interactive queries against it find nothing.

What shortens it:

• Standby replicas (numStandbyReplicas = 1): another instance keeps each stateful task’s stores current, and a failover restores only the tail it hadn’t applied.
• Persistent stores (persistentKeyValueStoreBuilder) checkpoint their changelog position; a task that comes back to the same machine restores only the tail after the checkpoint.
• A stable stateDir (a persistent volume) per instance: the assignor sees the checkpoints a restarted instance still has and gives it those tasks back. See Running in containers.
• Warmups instead of cold moves: with acceptableRecoveryLag tight, a task only moves to an instance that is caught up; a new or empty instance warms up first and takes over at a probing rebalance (probingRebalanceIntervalMs).
• Clean shutdowns (closeKafkaStreams): under exactly-once, a start after an unclean stop wipes local state and restores everything.
• More partitions: restore is per task, so smaller tasks restore faster and in parallel across instances.

import qualified Kafka.Streams.Config as C

cfg :: C.StreamsConfig
cfg = C.defaultStreamsConfig
  { C.applicationId       = "my-app"
  , C.bootstrapServers    = ["broker:9092"]
  , C.clientId            = "my-app-instance-1"
  , C.stateDir            = "/var/lib/my-app/state" -- on a persistent volume
  , C.numStreamThreads    = 4
  , C.numStandbyReplicas  = 1
  , C.processingGuarantee = C.ExactlyOnceV2
  }

Details: Scaling: standbys bound the failover time.

4. Listeners and progress

In the test driver, you see every record. In production, you have hundreds of thousands of records per second across multiple instances. Without visibility you can’t tell whether processing is keeping up, whether commits are succeeding, or whether an instance is stuck restoring.

What a running instance gives you:

Signal	Alert when
setStateListener	Any transition to StreamsError; StreamsRebalancing for longer than your restore budget
setStateRestoreListener	Restores getting slower (the changelog is growing)
streamsMetrics: process-total	The rate stops increasing while input lag grows
streamsMetrics: commit-aborted-total, task-recoveries-total	Any sustained rate
[streams] commit aborted log lines	High rate (something keeps failing)

Full list and patterns: Observability.

5. Orphan-topic detector

Deployments that rename operators (even accidentally) leave behind internal topics that nobody is using but everyone is paying for. A 100-partition changelog topic with 7-day retention consumes significant disk. Over months, these orphans accumulate. Detecting them at startup lets you clean up before they become a budget line item.

When you rename a stateful operator (or accidentally renumber its auto-generated name), the old internal topic on the broker becomes an orphan. It silently consumes disk forever.

detectOrphans :: Topology -> Text -> [TopicName] -> [OrphanInternalTopic] is a pure function. Call it on startup with your topology and an AdminClient.listTopics result:

topics  <- AdminClient.listTopics admin
orphans <- pure (detectOrphans topo appId topics)
forM_ orphans $ \o ->
  warn ("orphan internal topic: " <> unTopicName (orphanTopic o))

The detector only warns; it never deletes. Auto-deletion is a foot-gun (a misconfigured rollout would happily nuke live state). Make manual deletion an audited operator action.

Details: Observability: orphan internal topics.

6. Topology golden file

Code review can’t catch topology changes that accidentally shift auto-generated names. A seemingly innocent refactor (adding a map upstream of an aggregation) can renumber every subsequent operator, causing all internal topics to be recreated and state to be lost. A golden file test fails the build when the topology shape changes, forcing explicit review of the diff.

Snapshot the topology JSON into your repo:

import qualified Data.Aeson.Encode.Pretty as P
import qualified Kafka.Streams.Observability.Topology as Obs

writeGolden :: IO ()
writeGolden = do
  topo <- F.buildTopologyFrom myTopology
  BL.writeFile "test/golden/topology.json"
    (P.encodePretty (Obs.topologyDescription topo))

Add a test that fails if myTopology produces a different JSON. Even innocent-looking PRs surface the diff explicitly. Reviewing the diff is your last line of defence against the “I added a map and now my changelog topic has a different name” class of bug.

A diff touching sources, stores, or edges is a deploy you have to think about. A diff touching only processors (renumbered unnamed nodes) is harmless only if none of the renumbered nodes own state. See Topology evolution for the full classification.

7. Test rebalance and restart

The test driver runs everything in one thread with no failure injection. Production has network partitions, broker restarts, rolling deploys, and consumer group rebalances. These events trigger code paths your unit tests never exercise. Testing rebalance and restart scenarios before they happen in production reveals whether your state stores recover correctly, whether your external resources clean up properly, and whether your exactly-once guarantees hold under failure.

The test driver runs one task in one thread. Production runs multiple tasks across multiple instances. The gap between them is where the operationally-surprising bugs live.

Two extra tests every production topology should have:

• Rebalance test. Start two KafkaStreams instances in the same process, with the same applicationId, on one in-memory broker (Kafka.Streams.Runtime.MockBroker: newMockBroker, mockDriverFor, startKafkaStreamsWith). Produce input, close one instance mid-stream, and verify every input reached the output and the state the survivor restored is right.
• Restart test. Start a topology, feed some records, close it, start it again with the same applicationId and state directory. Verify the second run picks up from where the first left off and produces the right output. (If it doesn’t, your changelog or state-store wiring is wrong.)

Both tests catch problems that only manifest when the runtime is exercised in its full mode. This includes partition handoff and replay.

8. Do you need Riffle?

The base Kafka Streams library handles most workloads well, but it has limitations. External API calls block the stream thread. Exactly-once semantics only cover Kafka-to-Kafka pipelines. Riffle extensions address these gaps.

Riffle is the extensions tier. You can ignore it until a parity limitation bites. Common signs you’ll want it:

If you…	Reach for
Enrich from a slow external API (HTTP, gRPC)	Async I/O
Write to Postgres / S3 / Iceberg / HTTP and need EOS	Two-phase commit sinks
Join two streams that advance at very different rates	Watermark coordinator
Need to migrate a state store’s schema	SchemaVersioned store

Adopting Riffle is incremental. Each feature is a new module or a new smart constructor; selecting one doesn’t change anything else about your topology. See Riffle for the tour.

A minimal production config

Putting it together:

import qualified Kafka.Streams.Config as C

prodCfg :: C.StreamsConfig
prodCfg = C.defaultStreamsConfig
  { C.applicationId             = "my-app"
  , C.bootstrapServers          = ["broker-1:9092", "broker-2:9092"]
  , C.clientId                  = "my-app-instance-1"
  , C.numStreamThreads          = 4
  , C.numStandbyReplicas        = 1
  , C.processingGuarantee       = C.ExactlyOnceV2
  , C.commitIntervalMs          = Nothing            -- 100 ms under EOS
  , C.replicationFactor         = 3                  -- match your broker
  , C.stateDir                  = "/var/lib/my-app/state"
  }

Then the startup wiring:

main :: IO ()
main = do
  topo <- F.buildTopologyFrom myTopology

  -- Orphan-topic detector
  admin   <- AdminClient.newAdminClient (C.bootstrapServers prodCfg)
  topics  <- AdminClient.listTopics admin
  let orphans = detectOrphans topo (C.applicationId prodCfg) topics
  forM_ orphans $ \o ->
    warn ("orphan: " <> unTopicName (orphanTopic o))

  -- Construct + start
  valid <- either (fail . show) pure (validateTopology topo)
  ks <- newKafkaStreams prodCfg valid
  setStateListener ks $ \old new ->
    info ("state " <> show old <> " -> " <> show new)
  setStateRestoreListener ks defaultStateRestoreListener
    { onRestoreEnd = \tp store n -> publishRestore tp store n }
  startKafkaStreams ks

  -- Wait for shutdown signal
  awaitShutdown
  closeKafkaStreams ks

Recap

The eight things to set up before a production deploy. The trade-offs of each processingGuarantee. Why naming every stateful operator matters across deploys. What keeps restores short. The minimum listeners to wire up. The decision tree for whether to reach for Riffle.

Where to go from here

You’ve finished the tutorial. The rest of these docs are organised by what you’ll need them for:

If you’re…	Read
Designing a rolling deploy	Topology evolution
Sizing your cluster	Scaling and rebalancing
Deploying to Kubernetes or another container runtime	Running in containers
Writing to a non-Kafka sink with EOS	Exactly-once across systems
Building your observability	Observability
Trying to understand IQ semantics	Visibility versus ACID
On-call for a streams app	Runbooks
Enriching from an external system	Enrichment
Curious what auto-optimisation does	Topology optimization
Wondering what you can change at runtime	Dynamic topology changes
Looking at Riffle for the first time	Riffle: Flink-class extensions
Hitting an unfamiliar term	Glossary

The Overview page has the operator-facing map of everything.

Welcome to streams.