Kafka Streams Fundamentals

Data Engineering & Modern Data Platforms · Lesson 44 · Real-Time Data Engineering and Kafka

Lesson Overview

Finance does not want the five-minute activity tile from Structured Streaming with Spark repainted as revenue. They want a running total of purchase lines per course_id, in cents, that moves within a minute of the line landing on commerce.purchase_line. The morning job remains the number they close the books on. The running total is the number an instructor lead can glance at during a promotion without waiting for 06:00 UTC.

A Spark micro-batch can do that, and lesson 32 already owns a cluster. A second streaming engine needs a reason. The reason here is local and specific: the total is a per-key table, the input is already a Kafka topic, the output should be a Kafka topic other services can consume, and the team that owns commerce is prepared to run a JVM process beside the brokers. Kafka Streams is a library in that process, not a separate cluster. It reads topics, keeps state in stores backed by changelog topics, and writes topics. Apache Kafka ships it as org.apache.kafka:kafka-streams. There is no TechLearningPro Streams SDK, and there is no official Python implementation of this library. Faust and similar projects are not this API. The walkthrough is Java because that is the supported DSL.

/learn/kafka teaches the log the library consumes. This lesson teaches the topology a data engineer designs on top of it: which stream is a fact, which table is a dimension, why a key change costs a repartition, and when the Spark tile should stay the Spark tile. Windowed event time is the next lesson. This one aggregates a running total so the state story is visible without grace periods getting in the way.

The schema is the Avro subject from Schema Registry and Event Contracts, or the plain JSON from lesson 42 if the cutover has not happened. The topology names fields. It does not invent them. unit_price_cents stays integer cents. Refunds subtract. Deletes do not exist on this topic.

Learning Objectives

By the end of this lesson you will be able to:

1. Describe a Kafka Streams application as a library with a topology, an application.id, and tasks tied to input partitions.
2. Choose a KStream for purchase lines and a KTable for a course dimension and for the revenue result.
3. Explain why changing the key to course_id creates an internal repartition topic, and why that is acceptable here.
4. Join a rekeyed stream to a KTable only when the keys and partition counts match.
5. Say what a state store's changelog topic is for, and what a local RocksDB directory is not for.
6. Set processing.guarantee explicitly and state what exactly_once_v2 covers and what it does not.
7. Point the lake and the dashboards at the output topic, not at a store directory on a host.
8. Decide this running total belongs in Streams and the five-minute activity tile stays in Spark.
9. Test a topology with TopologyTestDriver without a broker.
10. Name the hot-key risk of aggregating by course_id and the partition-count limit on parallelism.

Prerequisites

• Kafka Topics, Partitions and Replication: tasks cannot outrun partitions, and a key decides which task sees a record.
• Kafka Delivery Semantics and Transactions: exactly_once_v2 uses the transactions that lesson explained. This lesson does not re-derive them.
• Lesson 43: the value schema, the key, and the rule that cents are cents.
• Lesson 32: the activity tile, its checkpoint, and its watermark. You should be able to say why this revenue total is a different product.
• A JDK that matches the Kafka release you run. Kafka 3.9 clients are a safe pin for the snippets if your brokers are 3.9. On a 4.x cluster, use the matching 4.x artifact. The DSL types below are the long-standing org.apache.kafka.streams API. Confirm deprecated methods in the release notes before copying older examples, especially anything that still calls TimeWindows.of without a grace.

Detailed Core Concepts

A topology is the job

You build a topology with StreamsBuilder, then you start a KafkaStreams instance on it. The builder records sources, processors, and sinks. At runtime the library splits the topology into tasks. Each task is assigned one or more partitions of the source topics and runs single-threaded. If commerce.purchase_line has six partitions, you get at most six tasks for that source, no matter how many hosts you start. Extra hosts wait. Fewer hosts run several tasks each. Scaling out means raising partition count first, with the key and ordering consequences from lesson 39, then starting more application instances with the same application.id.

application.id is the consumer group and the prefix of internal topics. Two different jobs must not share it. They would steal partitions from each other and write the same changelog names. One job's instances must share it. They would otherwise each consume every partition and double the total.

KStream and KTable

A KStream is a sequence of facts. Every purchase line is a new fact, including a refund. The stream does not collapse them. If the same business key is produced twice because Connect retried, both records are in the stream unless you dedupe on purpose.

A KTable is a current view of a key. A new record for a key replaces the previous view. The course dimension, commerce.courses, is a compacted topic keyed by course_id: title, active flag, and nothing else the total needs. The revenue result is also a table: one integer per course_id, updated as lines arrive. Modeling the result as a stream of "totals so far" without a table is how you emit an unbounded history of partial sums and call it a current number. You may still write each update to an output topic. The output is a changelog of the table. Consumers that want the current total keep the latest record per key. Consumers that want every revision can read the whole topic. Those are different readers, and the contract should say which one the dashboard is.

A GlobalKTable is replicated in full to every instance. It fits a small dimension you must join without co-partitioning. The course catalog for TechLearningPro can fit. This lesson still uses a KTable and a co-partitioned join, because that is the pattern you will be asked to operate when the dimension is no longer small. A global table is an optimization you choose after you can say why the partitioned one is inconvenient, not before.

Keys, repartition, and co-partitioning

Purchase lines are keyed by (order_id, line_id) so two lines do not overwrite each other. The total is keyed by course_id. After selectKey to course_id, records that must be aggregated together are not yet in the same partition. groupByKey marks the stream for repartition. Streams writes an internal topic, {application.id}-...-repartition, and reads it back. That topic is real Kafka: retention, replication, and disk. It is not a hint.

The cost is an extra hop and a new partition map. The benefit is a correct sum. The wrong fix is to change the source topic's key to course_id so the hop disappears. A popular course then dominates one partition of the extract, and two lines of one order can land on different partitions if you ever key them separately. Keep the source key as the line identity. Pay the repartition inside the job that needs a different key.

A KStream-KTable join requires the stream and the table to use the same key and the same number of partitions, and the same partitioner. The library throws or silently mis-joins when that is violated, depending on whether it can detect the mismatch at topology build. Do not disable the check. Repartition the stream to course_id, give commerce.courses and the repartition topic the same partition count, and join. The join result is "this line, plus the course row if we have one." An unknown course_id produces no join output for an inner join. That drops revenue on the floor. Use a left join and let a missing dimension pass the line with a null course, then count those nulls. The quality gate still owns referential integrity for the morning close. The live total should not hide lines because a catalog update is late.

State stores and changelogs

The aggregate's current totals live in a state store named in Materialized.as(...). The default persistent store is RocksDB, in a directory under the Streams state dir on the instance that owns the task. That directory is a cache of recovery, not the system of record. The system of record is the changelog topic, compacted, named from the application.id and the store name. On a restart or a rebalance, the task restores the store by reading the changelog.

If you delete the local directory and leave the changelog, the task restores. If you delete the changelog and leave the directory, a new instance comes up empty or from a partial copy and the total is wrong until you rebuild it from the source topic. If you delete both, you rebuild from commerce.purchase_line by resetting offsets, and you must accept a replay. Treat changelog topics like data topics: replication factor, min.insync.replicas, and a retention policy that does not delete the compacted history out from under a restoring task. Compacted topics still need segment.ms and a clean policy that matches Kafka's compaction rules. Someone setting a short retention.ms on a changelog "to save disk" will truncate a key that has not been updated recently and drop that course from the total after the next restore.

num.standby.replicas keeps extra copies of the store warmed on other instances so a rebalance restores less from the changelog. One standby is a reasonable production start when restore time has actually hurt. It is not a second source of truth. The changelog still is.

Interactive queries (KafkaStreams.store) let a process read the local store. They are a serving API for the instance that happens to host the key, with metadata to find that instance. They are not how the lake reads revenue. The lake reads the output topic through a sink you can replay. A dashboard that RPC-calls the Streams host will break the moment the task moves, and it will disagree with the warehouse without an offset anyone can name.

Processing guarantee

processing.guarantee defaults to at-least-once in the 3.x line: a crash can reprocess input and write the total twice as an update, and an output record can be duplicated. For a table keyed by course_id whose value is a sum, a duplicate input record changes the sum. At-least-once plus a retried source record double-counts cents. You already accepted that risk at the Connect boundary and pushed dedupe to the curated job. For the live total, decide explicitly.

exactly_once_v2 (the mode to name on current Kafka Streams; the older exactly_once and exactly_once_beta are the ones release notes tell you to leave) wraps the read-process-write in transactions so a rebalance does not publish a partial update, and so input offsets and output records commit together. It requires a cluster that supports transactions, which lesson 41 covered, and it shortens the commit interval because the default commit.interval.ms drops to 100 ms under that guarantee. It does not include the PostgreSQL insert. It does not dedupe a line_seq that Connect produced twice as two successful records with different offsets. Those are two inputs. Exactly-once processing will add both, correctly, once each. Idempotency on (order_id, line_id) inside the aggregate is still your code if duplicates can exist as distinct offsets. A simple way to see the limit: the transaction protects the topology's own outputs. It does not know the business key.

Set the config. Do not rely on a default you did not read for that artifact version.

Commit interval, cache, and freshness

The store cache batches writes to the changelog and to downstream nodes. A total can be updated in memory for up to commit.interval.ms before a consumer of the output topic sees it. For at-least-once, that interval defaults to 30 seconds. A product that asked for "about a minute" can live with 30 seconds. A product that asked for "a second" cannot, and lowering the interval increases write rate to the changelog. statestore.cache.max.bytes (the setting that replaced cache.max.bytes.buffering) bounds the cache. Turning the cache off makes every update visible sooner and makes the changelog hotter. Write the freshness number next to the config. The five-minute tile's freshness is a watermark question and is not this number.

When this is the wrong engine

Use Spark Structured Streaming for the activity tile: the sink is files, the team runs Spark, the watermark and the checkpoint are already the operational story, and the latency budget is the micro-batch. Use Streams when the input and the output are Kafka topics, the state is per key, and you can run and patch a JVM application with the brokers' client version. Use Flink, as lesson 33 framed it, when you need event-time timers, very large keyed state, and checkpoint alignment as the runtime's center. Do not stand up Streams, Spark, and Flink for three views of purchase lines. The running total is the Streams job. The tile stays in Spark. The books stay in the morning batch, which dedupes and applies the quality gate.

Architecture and Design Diagrams

mermaidCopy
flowchart LR
    subgraph inputs [Topics]
        lines[commerce.purchase_line key line id]
        courses[commerce.courses compacted key course_id]
    end

    subgraph app [Kafka Streams application.id course-revenue]
        rekey[selectKey course_id]
        repart[(repartition topic)]
        join[Left join course KTable]
        agg[Aggregate cents in state store]
        store[(RocksDB local)]
        change[(changelog compacted)]
    end

    subgraph out [Consumers]
        topic[commerce.course_revenue]
        dash[Dashboard consumer latest per key]
        lake[Lake consumer replay]
    end

    lines --> rekey --> repart --> join
    courses --> join
    join --> agg
    agg <--> store
    agg --> change
    agg --> topic
    topic --> dash
    topic --> lake

The local RocksDB files are disposable. The changelog and commerce.purchase_line are what a restore uses. The lake does not read RocksDB.

mermaidCopy
sequenceDiagram
    participant Line as purchase_line
    participant Task as Streams task
    participant Store as State store
    participant Log as Changelog
    participant Out as course_revenue

    Line->>Task: Line keyed by order and line id
    Note over Task: Repartition by course_id
    Task->>Store: Read current cents for course_id
    Task->>Store: Add signed unit_price_cents times quantity
    Task->>Log: Write new total on commit
    Task->>Out: Write new total on commit
    Note over Task,Log: Crash before commit replays the line
    Note over Out: exactly_once_v2 does not publish the uncommitted total

A second instance with the same application.id takes a different subset of partitions. It does not recompute the same course_id unless you broke co-partitioning and the key is on two tasks, which would be a bug.

Technical Explanation

Signed amount

The aggregate adds unit_price_cents * quantity for sale and subtracts for refund. The result is a long. Overflow is unlikely at catalog prices and fatal if it happens silently in two's complement. The aggregator checks the sign of the line type and rejects any other type to a side topic you can name, commerce.course_revenue.rejected, produced by a branch in the topology. That side topic is an application sink, not Connect's dead-letter queue. Do not catch and add zero. A zero hides a producer bug inside a total that still looks plausible.

Unknown course_id values from the left join still add into the total. The output record carries dimension_missing=true when the table had no row. The dashboard can show the total and a count of missing dimensions. Dropping the line would make the live number smaller than the morning number for a reason nobody sees.

Internal topic names

Expect topics whose names start with the application.id. Repartition topics are not compacted in the same way changelogs are; they are a stream of rekeyed records and they need retention long enough to cover a restore that has not finished reading them. Changelog topics are compacted. Do not point a Connect sink at "every topic" with a regex that includes internal topics, or you will land repartition noise in the lake and, worse, a sink consumer in a group that does not belong there. Name the output topic explicitly.

Reset and rebuild

KafkaStreams.cleanUp() deletes local state. The application, on the next start, restores from changelogs if offsets say so. The application reset tool (kafka-streams-application-reset) is the supported way to rewind input offsets and delete internal topics when you intend a rebuild. Running it against the wrong application.id deletes another job's state. A rebuild double-counts if any downstream system treated the output topic as an append-only ledger of deltas rather than as a table of current totals. The output contract for this job is current total per course, so a rebuild rewrites keys. Consumers that overwrite by key survive. Consumers that sum the output topic will double. Write that sentence in the topic contract from lesson 43.

Testing without a broker

kafka-streams-test-utils provides TopologyTestDriver. You pipe input records with timestamps and you read the output topic. No broker, no RocksDB durability, and no test of exactly_once_v2. The driver is for topology logic: sale adds, refund subtracts, unknown line type does not add, a second line for the same course accumulates. An integration test on a real cluster is still required before you trust commit semantics and rebalance. The driver is how you keep the cents logic honest in CI.

Serialization

Use the same Avro serde as the subject, or a JSON serde that matches lesson 42's bare JSON, on the source. Use a primitive long serde for the aggregated cents so the output topic is boring. Do not log the full line at info. The line is not a payment token, and it is still commercial data. The output record is course_id, net_cents, line_count, dimension_missing, updated_event_time. updated_event_time is the max event time applied, which lesson 45 and lesson 47 will want. It is not stream time and it is not wall clock. Computing it in the aggregate is a few lines and it saves a second job later.

Practical Implementation or Design Walkthrough

Problem

Maintain net_cents per course_id from commerce.purchase_line, joined to commerce.courses so a missing dimension is visible. Publish commerce.course_revenue. Keep the Spark activity tile alone.

Topology

The types below are the Java DSL. Serde setup is omitted where it is the registry serde from lesson 43; the aggregate's logic is the part that decides money.

javaCopy
StreamsBuilder builder = new StreamsBuilder();

KStream<LineKey, PurchaseLine> lines = builder.stream(
    "commerce.purchase_line",
    Consumed.with(lineKeySerde, purchaseLineSerde));

KTable<String, Course> courses = builder.table(
    "commerce.courses",
    Consumed.with(Serdes.String(), courseSerde));

// Source key is the line. The total's key is the course.
// groupByKey after selectKey creates the repartition topic.
KStream<String, PurchaseLine> byCourse = lines.selectKey(
    (lineKey, line) -> line.courseId());

KStream<String, LineWithCourse> joined = byCourse.leftJoin(
    courses,
    (line, course) -> new LineWithCourse(line, course),
    Joined.with(Serdes.String(), purchaseLineSerde, courseSerde));

KTable<String, Revenue> revenue = joined
    .groupByKey(Grouped.with(Serdes.String(), lineWithCourseSerde))
    .aggregate(
        Revenue::zero,
        (courseId, line, total) -> total.apply(line),
        Materialized.<String, Revenue, KeyValueStore<Bytes, byte[]>>as("course-revenue")
            .withKeySerde(Serdes.String())
            .withValueSerde(revenueSerde));

revenue.toStream().to(
    "commerce.course_revenue",
    Produced.with(Serdes.String(), revenueSerde));

Revenue.apply adds signed cents for sale and refund and refuses any other line_type by recording a rejection counter in the value rather than throwing, if you want the task to stay up. Throwing fails the task, which is the right call when an unknown type means the schema contract broke. Prefer failing the task if lesson 43's enum says the type cannot happen. A poison record then blocks the partition, which is loud. A side topic is the alternative if you must keep the total moving. Pick one in the contract. This walkthrough fails the record by throwing on an unknown type so the partition stops and the schema owner is paged. Do not swallow it.

commerce.courses and the repartition of commerce.purchase_line need the same partition count. Create both with that count before the first start. A later change to partition count is a rebuild, not a config tweak.

Config that must be explicit

propertiesCopy
application.id=course-revenue
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
processing.guarantee=exactly_once_v2
commit.interval.ms=1000
num.standby.replicas=1
default.deserialization.exception.handler=org.apache.kafka.streams.errors.LogAndFailExceptionHandler

LogAndFailExceptionHandler stops the instance on a payload that does not match the serde. LogAndContinueExceptionHandler skips the record and moves the offset. Skipping a purchase line understates revenue and looks like a healthy consumer. Fail closed here. The Connect dead-letter topic from lesson 42 is where sink conversion failures already go. This job is a different failure domain: bad bytes on the source partition stop that partition's task.

commit.interval.ms=1000 is a freshness choice under exactly_once_v2, whose own default is already short. One second matches a "within a minute" product with room for the JDBC poll interval from lesson 42, which is the larger delay. Do not tune Streams to 100 ms while the source only polls once a minute and then claim the total is real-time from the database.

TopologyTestDriver

javaCopy
try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
    TestInputTopic<LineKey, PurchaseLine> input = driver.createInputTopic(
        "commerce.purchase_line",
        lineKeySerde.serializer(),
        purchaseLineSerde.serializer());
    TestOutputTopic<String, Revenue> output = driver.createOutputTopic(
        "commerce.course_revenue",
        Serdes.String().deserializer(),
        revenueSerde.deserializer());

    input.pipeInput(lineKey("o1", "l1"), sale("course-a", 1500, 1));
    input.pipeInput(lineKey("o2", "l1"), refund("course-a", 1500, 1));

    Revenue last = output.readKeyValuesToMap().get("course-a");
    // 1500 in, 1500 out, net zero, two lines applied
    assert last.netCents() == 0L;
    assert last.lineCount() == 2L;
}

The driver lives in org.apache.kafka:kafka-streams-test-utils, same version as kafka-streams. Add a third record with line_type outside the enum only if your serde can represent it; with a strict Avro enum it will not deserialize, which is the fail-closed path. Test that path with a byte array that is not Avro and expect the handler you configured, in an integration test. In the driver, call Revenue.apply directly for the unknown-type unit test so the money rule is covered even when the serde cannot express the bad value.

Operate

Start two instances with the same application.id if you have enough partitions to matter. Watch restore time by restarting one instance and reading the log line that reports changelog restoration. Confirm the output topic has one key per course that has ever had a line, and that a restart does not append a second independent total. Confirm a consumer that sums the output topic gets a nonsense number, and write that into the consumer's README so the next dashboard does not do it.

Rebuild procedure, for the runbook: stop instances, run the application reset tool for course-revenue only, start instances, let them reprocess commerce.purchase_line from the earliest offset you still have, and compare net_cents for one course against a SQL sum of the raw prefix for that course. They will disagree by lines still in the JDBC poll interval and by duplicates the SQL dedupes. The comparison is on deduped (order_id, line_id), not on raw file row counts.

Enterprise Real-World Scenarios

Promotion-day revenue by course

A promotion doubles line volume for three hours. Partition count on commerce.purchase_line is 6, keyed by line identity, so the extract stays balanced. The repartition by course_id is not balanced: two courses take most of the lines, and two tasks sit at high CPU while four are bored. The total stays correct. Lag on those two tasks grows by seconds, still inside the one-minute budget, because the aggregate is a long add and not a join to the learner table. The failure mode to refuse is "fix the skew by randomizing the revenue key." That makes the total wrong. The failure mode to plan is "if one course exceeds a task's budget, split the product": publish per-course totals as now, and a separate promotion leaderboard that is allowed to be approximate. Do not partition the financial total by a salt. The morning SQL remains the close, deduped, with the quality gate. The Streams topic is the intra-day view. Security is the same subject as lesson 43: the output has cents and course ids, not purchaser identifiers. Leave learner_id off this topology even if it appears on a future line schema. The total does not need it, and adding it to the store key would explode state.

A dimension update during the event

Course course-a is renamed in commerce.courses. The KTable updates. The next purchase line joins to the new name. Lines already aggregated do not rewrite their historical output unless you reprocess, and the output record stores course_id, not the title, so a rename does not change net_cents. A team asked to "recompute history on every dimension change." That is a replay of commerce.purchase_line, which you can do, and it is not what a table join does by default. The join is current-dimension-at-event-arrival, with the usual stream-table timing: a line that arrives before the course row is dimension_missing=true until you decide to reprocess or to buffer. Buffering in a custom store is how this small topology becomes unoperable. Emit the flag. Let the catalog stream catch up. Let the morning job, which joins a pinned dimension the way lesson 17 described, be the place where a missing course fails a curated publish.

Best Practices

• One application.id per job. Put the id in the runbook next to the changelog prefix.
• Keep source keys on the business identity of the event. Repartition inside the job that needs a different key.
• Match partition counts before a stream-table join. Do not catch the exception and continue.
• Fail deserialization on financial topics. Skip-and-continue is a silent revenue bug.
• Treat changelogs as production topics. Do not put a short retention on them.
• Point sinks, lakes, and dashboards at the output topic. Read the state store from a test or from interactive queries you have actually designed, not from a shell on the host.
• Set processing.guarantee in the config you review. Name exactly_once_v2 when you need it, and still dedupe business keys that can be duplicated upstream.
• Put updated_event_time on the output so freshness is a field, not a guess from wall clock.
• Test money in TopologyTestDriver. Test rebalance on a real cluster before production.
• Leave the activity tile in Spark. Leave the books in the batch dedupe. Write the boundary in the service README.
• Size partitions for the heaviest course_id you have seen, not for the average.
• Pin kafka-streams to the broker generation you run.

Common Mistakes and Anti-Patterns

Sharing application.id across jobs. They are one consumer group and one changelog namespace. Totals mix. One id, one topology.

Keying the source topic by course_id to avoid a repartition. The extract's partitions skew, and line identity is gone. Repartition in the job.

Inner join to the dimension. Lines for a course the table has not ingested yet disappear from the live total. Left join and flag the miss.

Summing the output topic. Each record is the new total, not a delta. A sum double-counts every update. Consume by key and keep the latest, or read the store API you designed. The contract should say "current total."

Local disk as the source of truth. An instance dies, the volume goes with it, and someone is surprised the changelog restore takes minutes. The surprise means the changelog was never operated. Restore is the design, not the failure.

LogAndContinueExceptionHandler on this job. A bad record is skipped, the offset commits, and the course is short by that line until a human notices a reconciliation miss. Fail the task.

Exactly-once as a synonym for "the business key is unique." exactly_once_v2 avoids publishing a torn transaction from this process. Two source records with different offsets and the same (order_id, line_id) are both applied. Dedupe if the source can do that, or accept that the morning job is the deduped close.

A transform, a Streams app, and a Spark job all computing "revenue." Three definitions. One topology owns the live total. The batch owns the close. The tile is not revenue.

Interactive queries as the lake integration. The lake cannot replay a socket. Write the topic.

Troubleshooting and Design Review

Symptom: After a rebalance, totals drop to zero and climb back over several minutes.

Likely cause: restore from changelog, and the standby was 0, or the changelog is large because the cache was disabled and every line is a changelog record.

Investigation: task logs for restore, changelog end offsets, num.standby.replicas.

Corrective action: if the climb matches the source, wait, and add a standby if the minutes exceed the product budget. If the climb stabilizes below the SQL sum, the changelog was truncated. Rebuild from the source topic with the reset tool and fix retention.

Symptom: One task's lag grows during the promotion. The other tasks are idle.

Likely cause: course_id skew after the repartition. The total is still correct and slow for the hot courses.

Investigation: record rate per task, not only the consumer-group lag average. Averages hide the hot partition.

Corrective action: do not salt the financial key. Confirm the budget. If the hot course exceeds it, split the product or give the topology more time. Adding instances without adding partitions does nothing.

Symptom: The output topic's sum is about twice the morning SQL.

Likely cause: a consumer summed changelog-style updates, or a rebuild replayed without truncating the previous output, and the consumer summed that too.

Investigation: one course_id, last record versus sum of all records for that key.

Corrective action: fix the consumer to take the latest value. If you need deltas, emit deltas under a different name and a different contract. Do not make the current-total topic do both.

Symptom: Unknown line_type and the task is in ERROR, partitions unassigned.

Likely cause: fail-closed handler, which is what you configured. A producer broke the contract.

Investigation: the exception's offset. Read that offset. Do not skip it in the handler as a mitigation that becomes permanent.

Corrective action: fix or quarantine the producer record, then start the instance. If you must move past one poison record, do it as a one-time reset of that offset with a written revenue adjustment, not by changing the handler to continue.

Symptom: Dimension flags are true for every line.

Likely cause: commerce.courses is empty, or its key serde does not match course_id, or the partition counts differ and the join is not actually co-partitioned.

Investigation: record count on the dimension topic, a sample key, partition counts on both sides of the join.

Corrective action: fix the dimension producer or the partition count. Reprocess if you need the flags rewritten. A global table would have hidden the partition bug and is not the first fix.

A design review asks: What is the source key, the aggregation key, and the repartition topic? What is application.id? Where is the changelog and what is its retention? Latest-value or delta on the output? What does a duplicate (order_id, line_id) do to net_cents? What happens on a serde error? Why is this not the Spark tile? What will you compare against SQL after a rebuild?

Interview Questions and Answers

Q1. What is a Kafka Streams application, physically?

A JVM process using the kafka-streams library. It joins the consumer group named by application.id, is assigned tasks that correspond to input partitions, stores state locally, and logs that state to compacted changelog topics. There is no separate Streams cluster to deploy. Parallelism cannot exceed the partition count of the inputs. I run more instances only after the topic has the partitions to feed them.

Q2. When do you model something as a KStream, and when as a KTable?

A purchase line is a fact that should not replace a different line, so it is a KStream, even if Connect occasionally retries one line. The course catalog is the current row per course_id, a KTable on a compacted topic. The revenue total is a KTable because the product wants the current cents, not the history of partial sums, even though that table's changelog is itself a stream of updates. I write the output contract as "latest record per course wins" so nobody sums it.

Q3. Why does selectKey to course_id followed by groupByKey have a cost?

Records with the same course_id are not in the same partition while the source is keyed by line identity. The aggregation is only correct if they meet. Streams writes a repartition topic and consumes it. That is extra disk, extra lag, and a topic you must not sweep into a generic sink. I still do it. Changing the source key to course_id would skew the extract and throw away line identity, which other consumers need.

Q4. What has to be true for the line-to-course join to return the right course?

The stream must be keyed by course_id, the table must be keyed by course_id, both sides must have the same partition count, and they must use the same partitioner. I left-join so a missing dimension is a flag, not a vanished line. An inner join would understate revenue whenever the catalog is late. I do not join to a large enrollment fact in this topology. That is a different state budget.

Q5. What is the changelog for, and what is the RocksDB directory for?

The changelog is the replicated log of store updates. The local directory is a recoverable cache so the task does not read the entire changelog on every start. Restore reads the changelog. Deleting local state is an inconvenience. Deleting or truncating the changelog loses totals for keys that are not rebuilt from the source. I operate changelog topics with the same replication standards as the purchase topic, and I do not set a short retention on them.

Q6. What does exactly_once_v2 guarantee here, and what does it ignore?

It commits input offsets and output records transactionally so a crash does not publish a total the input offset does not cover, and so a rebalance does not duplicate those outputs. It does not wrap the database insert that created the line. It does not collapse two Kafka records that are duplicates at (order_id, line_id) with different offsets. I still compare a rebuilt total to a deduped SQL sum. Lesson 41 is the transaction mechanism. This config turns it on for the topology. It is not a business-key deduper.

Q7. Why is the deserialization handler fail-close?

LogAndContinueExceptionHandler commits past a payload the serde rejected. That line never enters net_cents. The task stays up and the total is quietly short. On this topic I use LogAndFailExceptionHandler so the partition stops on bad bytes. A one-off skip is an incident decision with an adjustment, not a config change that survives into next quarter.

Q8. How do you test the aggregate without a broker, and what does that test not prove?

TopologyTestDriver and TestInputTopic feed records through the topology and read commerce.course_revenue. I assert sale, refund, and accumulation. The driver does not run exactly_once_v2, does not rebalance, and does not restore a changelog. Those belong in an integration test on a real cluster before production. I do not treat a green driver test as a rebalance test.

Q9. Why not implement this total in the Spark job that already reads Kafka?

The activity tile is a windowed event-time count with a file sink and a checkpoint, owned by the Spark platform. The revenue total is a current value per course, output to a topic, with a team that can run a JVM service. Putting both in one micro-batch couples their release and their failure domain. I would put the total in Spark if there were no Streams owners and the sink could be a table update. I would not introduce Streams for the tile. Two products, two engines that already fit them, one batch close underneath.

Q10. A hot course makes one task late during a promotion. What do you change?

I confirm correctness first. Skew makes the hot key slow and the total right. I do not salt course_id on a financial total, because the parts would need a second exact combine and the live key would no longer be the course. I add partitions only if the skew is across many courses and I am actually out of parallelism. For one dominant course I either accept the lag if it is inside the minute, or I split a non-financial leaderboard off this topology. I compare the final totals to deduped SQL after the promotion.

Q11. How should a lake consumer use commerce.course_revenue?

As a changelog of current totals: keep the latest record per course_id, and use updated_event_time as the event-time freshness of that total. Do not append every update into a sum. The morning purchase fact, deduped on the line key, remains the close. The Streams topic is the intra-day view. Module 5 can land the topic into a table that merges on course_id. It should not land it as an append-only fact of partial sums.

Hands-On Exercises

Exercise: A course total you can rebuild

Problem statement. Implement the running total, prove refunds subtract, and prove a rebuild from the source topic reaches the same net_cents for a fixed input.

Tasks.

1. Define Revenue.apply for sale, refund, and unknown type. Unit-test it without Streams.
2. Build the topology: rekey, left join to a course table, aggregate, write commerce.course_revenue.
3. Drive two lines and one refund through TopologyTestDriver. Assert net_cents and line_count. Include a line whose course is absent and assert the missing-dimension flag.
4. On a real cluster, run the app with processing.guarantee=exactly_once_v2 and two partitions or more. Produce the same lines. Restart an instance and show the total is unchanged.
5. Reset the application with the official reset tool and reprocess. Compare totals with a SQL or spreadsheet sum of the same lines. Explain any duplicate-key behavior in a sentence.
6. Write the output contract: latest value per course_id, not a delta. Name one consumer mistake that doubles the money.
7. State why you did not put this logic in the Spark activity tile.

Expected deliverables. The topology, the driver test, the before-and-after restart totals, the rebuild comparison, and the contract paragraph.

Evaluation rubric.

You pass when a refund brings the driver total back down, a restart does not double it, and the contract forbids summing the output topic.

Lesson Summary

Kafka Streams is a library that turns topics into a topology of tasks, one task per input partition, with state in local stores restored from changelog topics. Purchase lines stay a KStream keyed by line identity. The revenue total is a KTable keyed by course_id, reached through an explicit repartition. The course dimension is a KTable left-joined so a late catalog row does not drop cents. application.id names the group and the internal topics. exactly_once_v2 makes the topology's own writes transactional and does not dedupe upstream business keys. Fail deserialization. Point the lake at the output topic and read it as the latest total per course. Keep the five-minute activity tile in Spark and the financial close in the deduped morning job.

Event Time, Windows and Late Data puts a clock on a topology. The running total deliberately ignored lateness. The activity tile cannot.

Further Reading

• Kafka Streams core concepts — tasks, application.id, state, and time. Match the page to your release; 4.x moved some defaults.
• Kafka Streams DSL — KStream, KTable, joins, and Materialized.
• Configuring a Streams application — processing.guarantee, commit.interval.ms, num.standby.replicas, and the deserialization handlers.
• Application reset tool — the supported rebuild path. Do not delete internal topics by hand.
• Data Delivery Semantics and Transactions — the transactions exactly_once_v2 relies on.
• /learn/kafka — partitions and keys, which set this job's parallelism.
• Structured Streaming with Spark — the tile this lesson did not replace.