In this lab, you will build one Kafka Streams topology that filters and routes order events, counts orders per store in a state store, counts them again in one-minute event-time windows with grace, and enriches orders from a status table. You will then inspect the internal topics it created, kill an instance and watch another restore from the changelog, and crash the application mid-batch under exactly_once_v2 and under at_least_once to see where the two guarantees part.

Every feed run publishes the same 300 events, ord-1000 through ord-1299, over six stores store-17 through store-22, with every tenth event marked CANCELLED: 270 placed orders, 50 of them for store-17, and 115 at or above the high-value threshold. Each event carries an occurredAt timestamp in epoch milliseconds, with 200 ms between events. Choose a base time exactly on a UTC minute boundary so all 300 timestamps fit in one window. Compare observed counts with these expected values, then investigate differences in input history, timing, or recovery.

The broker is environment E1: the single kafka-local container running apache/kafka:4.3.1, reachable from the host at localhost:9092, with every command-line tool run inside it through docker exec. The Java code is environment E3 and runs on the host with Java 21 and Maven; start from the producer project layout from the producers module and add the streams dependency shown in Setup. The image's single-node configuration sets transaction.state.log.replication.factor and transaction.state.log.min.isr to 1, which Phase 6 needs.

Copy the producer project and rename it:

In pom.xml, set <artifactId> to windowed-order-counter, configure the Exec Maven plugin with <mainClass>${exec.mainClass}</mainClass>, and set <exec.mainClass>com.example.kafka.OrderCounter</exec.mainClass> inside the POM's <properties>. This lets -Dexec.mainClass select OrderFeed in later commands. Add the Streams library next to the existing kafka-clients dependency:

Run this lab against a fresh E1 broker with no earlier topics or application state. If kafka-local already contains data you need, use a separate lab environment. Run the following command when the container name and port are free, then create the five stream topics with three partitions each:

The status table's source topic needs the same three partitions, because a stream-table join requires co-partitioned inputs, and compaction, because each order id has one current row:

Save OrderJson.java, a builder and field reader for the fixed five-field value, so the lab needs no JSON library:

Save OrderFeed.java. Its only argument is the base instant; event i gets occurredAt of base plus i * 200 ms:

Save OccurredAtExtractor.java. Without it, Streams would window on the producer-assigned record timestamp under the default CreateTime policy, the moment the feed ran rather than the moment the customer placed the order:

Save LoggingRestoreListener.java: a class implementing org.apache.kafka.streams.processor.StateRestoreListener whose constructor takes the instance name. Its onRestoreStart prints one line beginning restore start with the instance, store name, TopicPartition, starting offset, and ending offset; its onRestoreEnd prints a line beginning restore end with the store, partition, and total restored; its onBatchRestored does nothing.

Save OrderCounter.java. The first argument names the instance and its state directory. Further key=value arguments become Streams properties, except lab.crash.after.store17 and lab.output.suffix. Phase 6 uses these lab options to halt the JVM inside the downstream sub-topology and isolate each run's output topics. The task observer prints current assignments every five seconds; the first record for lines only describe earlier activity. The partition calculation assumes this topology's default string-key partitioning and two-partition repartition topic.

Counts become strings before the sinks so the console consumer prints them without a deserializer flag. The one-second commit interval flushes cached results quickly; the at_least_once default is 30 seconds.

Compile once, start instance a in Terminal A, and save the printed topology as observations/topology.txt:

From Terminal B, publish the feed:

Dump the high-value output:

What you should see: the description lists two sub-topologies. Sub-topology 0 has two sources, orders.placed and orders.status, and three sinks, one writing to orders-by-store-repartition, which the broker holds as order-counter-orders-by-store-repartition; sub-topology 1 has that topic as its only source and two sinks. phase-1.txt has exactly 115 lines, each with a store- key and an amount of 105000, 130000, or 155000. Terminal A prints six first record for lines, because instance a owns both downstream tasks.

Dump the running count and reduce it to the last value per key:

List every topic with kafka-topics.sh --list, then run --describe on order-counter-orders-by-store-repartition, order-counter-orders-per-store-changelog, and order-counter-orders-per-store-minute-changelog. Append the list and the three descriptions to observations/phase-2.txt.

What you should see: the last values are 50, 40, 50, 40, 50, 40 for store-17 through store-22. Caching may combine intermediate updates, so grade the final values rather than the number of output lines. The list shows one -repartition topic, changelogs for orders-per-store and orders-per-store-minute, and, since topology optimization is off by default, a changelog for the order-status source table. The repartition topic has two partitions, not three, with cleanup.policy=delete and unlimited retention.ms. The orders-per-store changelog is compact; the windowed changelog is compact,delete with a retention.ms of at least the window size plus grace. Record the partition counts and configs for Q1 and Q3.

Start a consumer on the windowed output in Terminal B without --from-beginning, so only new updates appear:

Send four store-17 orders one at a time from Terminal C. Before sending the next, wait for the expected output or a skipped-record log entry. Use a documented timeout when diagnosing missing output. All four use this command with a different key and value; the first is:

Copy the consumer output and the relevant Terminal A lines into observations/phase-3.txt, then stop the Terminal B consumer with Ctrl-C.

What you should see: ord-2001 prints store-17|2026-09-12T10:01:00Z with count 1. ord-2002 prints store-17|2026-09-12T10:00:00Z with count 51, the feed's 50 plus this late order: a revision of an earlier result under the same key. ord-2003 prints the 10:03:00Z window with count 1. ord-2004 prints nothing on the windowed topic, and Terminal A logs a warning about skipping a record for an expired window that names the record timestamp and the stream time. The running count for store-17 still reaches 54, because a count without a window never closes.

Publish a status row and wait for table applied ord-1042=PAID in the application log before sending the next order. A stream-table join looks up table state at processing time, so this order matters:

Send ord-1042 as a store-17 order with amount 5000 and occurredAt 1789207410000 using the Phase 3 producer command. Wait for joined ord-1042=PAID|... in the application log before changing the table. Publish ord-1042|SHIPPED, wait for table applied ord-1042=SHIPPED, and send another ord-1042 event at 1789207420000. Wait for joined ord-1042=SHIPPED|... before continuing. Finally send ord-2005 for store-17 at 1789207430000 without ever publishing a status row for it. Read analytics.orders-with-status from the beginning with print.key=true and save it as observations/phase-4.txt.

What you should see: exactly two lines, both keyed ord-1042, the first value starting PAID| and the second SHIPPED|. Publishing SHIPPED did not re-emit the first joined record, and ord-2005 produced nothing because an inner join has no row to match. The running count for store-17 is now 57.

Leave instance a running and start instance b from Terminal B:

After the rebalance, publish a second feed with a later base so its records land in an open window:

After both instances report RUNNING, use the latest active task lines to find the owner of the order-counter-orders-by-store-repartition partition printed for store-17. Wait until the running count reaches 107 and kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group order-counter reports zero lag for the input and repartition partitions. Run that command inside kafka-local through docker exec, as in earlier phases. Save the assignments, committed progress, and count. Set APP_PID to the current owner's actual startup pid in the terminal where you run the next command, then stop that process:

Wait for the survivor to take over and measure the elapsed time. Total recovery includes failure detection, rebalance, and any required state restoration. Send one more store-17 order, ord-2006 at 1789211100000, and read the running count again. Then stop the survivor with Ctrl-C, restart it unchanged, and note any restore lines. Stop it again, delete only its own folder under the project's state directory, and restart it. Save all restore lines and counts as observations/phase-5.txt.

What you should see: a new owner restores any missing local state from the available changelog history. Record the actual starting and ending offsets; retention and compaction mean a restore need not start at offset 0. After takeover, ord-2006 brings the running store-17 count to 108. A clean restart can reuse local state and may require little or no replay. Deleting the state directory forces recovery from Kafka for the stores the instance owns. Compare the listener output for all three cases instead of assuming a fixed restore range or duration.

Stop every instance. Create separate output topics for the two fresh application ids so earlier output cannot hide missing results. Use each application id and state directory only for its crash attempt and restart:

Start the exactly-once run. It reads orders.placed from the beginning and halts when its twentieth store-17 record reaches the downstream sub-topology, before counting that record:

After it halts, run the same command without the crash argument. Keep lab.output.suffix=.eos. Wait until the application reports RUNNING and the consumer-group description for order-counter-eos reports zero lag on input and repartition partitions. Then dump the input and committed output:

Reduce phase-6-eos.txt to the last value per key with the Phase 2 awk command. Stop the exactly-once instance, then start the at-least-once crash run:

Restart with the same arguments except lab.crash.after.store17=20. Wait for RUNNING and zero lag for order-counter-alos, allowing its 30-second commit interval to elapse. Repeat the output dump against analytics.orders-per-store.alos, saving observations/phase-6-alos.txt, and reduce that file separately. Put the two comparisons and the observed restore offsets in observations/phase-6.txt.

What you should see: both crash runs print the halt message and exit. After exactly-once recovery finishes, each final per-store count equals the input count. Record the actual restoration range; do not assume offset 0. Under at-least-once processing, replay may repeat upstream publications and inflate a downstream count, but this crash trigger does not guarantee an overcount. An equal result is also valid evidence for this run, not proof of an exactly-once guarantee. Under exactly-once processing, the downstream stage can read only committed repartition records. Those upstream writes remain committed even if a later downstream transaction aborts. Recovery excludes the downstream transaction's aborted state and output updates and reprocesses its input. There is no single transaction covering the whole topology at once.

Stop every instance with Ctrl-C, then stop the broker and remove the container and its data:

Remove the local state directories:

Verify that nothing is left:

The output should contain only the header line, and ls state should report that the directory does not exist. Keep the source and observations for your report.

You built one topology and let its internal topics show how Kafka Streams organizes work. The map produced a new key-value pair for each record, but only the repartition topic brought every store-17 record to the task that owned its count, which split the topology into two sub-topologies with different partition counts. The changelogs held replacement counts, so compaction was appropriate. The repartition topic held individual orders that each needed to contribute to the count, so compaction would have lost input. The windowed count showed that membership comes from event time, that a late record inside grace revises an earlier result under the same key, and that the aggregation drops a record that arrives after the window closes while an unwindowed count still accepts it. The join showed a lookup against current table state rather than a buffered match.

The failure phases tied state to recovery. A task can move to another instance, which restores missing state from the available changelog history. A clean restart can reuse local files; deleting them requires more recovery work. Under exactly_once_v2, each stream-thread transaction commits its output, recoverable state updates, and consumed offsets together. Under at_least_once, a replay can publish a record twice and inflate the downstream count, although a particular crash may not expose that window. Console output and duplicate business events remain outside the processing guarantee.