Stream Processing and Kafka Streams

Stream Processing Is Continuous Computation Over Data That Never Ends

Batch processing operates on a bounded dataset: a file, a table snapshot, a day's worth of rows extracted from a warehouse. The job starts, reads everything that exists as of that moment, computes a result, and finishes. The defining property is that "all the data" is a knowable, fixed set at the moment the job runs — yesterday's orders table has an exact row count, and a batch job can process every row and then simply be done.

Stream processing operates on an unbounded stream: events keep arriving indefinitely, and there is no moment at which "all the data" exists, because more of it is always still coming. A stream processing job does not run once and finish — it runs continuously, computing and updating results as each new event arrives, for as long as the application is deployed. The question a stream processing job answers is not "what is the total for today" computed once at the end of the day, but "what is the running total right now" continuously, updated with every new event.

Kafka topics are naturally unbounded logs — a topic never "finishes" the way a file does — which is why Kafka is the dominant storage substrate stream processing systems are built against. A stream processing engine reads continuously from one or more topics, applies transformations, aggregations, and joins as data arrives, and typically writes its output back to other Kafka topics, which downstream consumers or other stream processing jobs can read from in turn.

Kafka Streams Is a Client Library, Not a Separate Cluster

This is the single most important architectural fact about Kafka Streams, and the thing that distinguishes it from both Kafka Connect (Module 13) and from external stream processing frameworks like Apache Flink or Spark Structured Streaming. Kafka Streams is a Java library that you add as a dependency to your own application. There is no separate "Kafka Streams cluster" to deploy or operate — a Kafka Streams application is simply your own JVM process, running your own business logic, that happens to use the Kafka Streams library to read from and write to Kafka topics with stream processing semantics built in.

Kafka Connect:
  You submit a JSON config to a Connect cluster (separate worker
  processes you or your platform team operate). Your code is the
  configuration; the connector plugin does the work.

External engine (Flink, Spark Structured Streaming):
  You submit a job to a separately-operated cluster (JobManager /
  TaskManagers, or a Spark cluster). Your job runs on infrastructure
  someone deploys and scales independently of your application.

Kafka Streams:
  You write a normal application (say, a Java service, or a
  Dockerized JVM process) and add the kafka-streams library as a
  dependency. YOUR application IS the stream processing job. You
  deploy it exactly like any other service -- Kubernetes, ECS,
  a JAR on a VM -- and scale it exactly like any other service.

This has real operational consequences. A Kafka Streams application does not need a separate operations team running a stream processing cluster — the same deployment, monitoring, and on-call practices your organization already uses for any other service apply directly. Scaling out a Kafka Streams application means running more instances of your own application, each of which automatically claims a share of the input topic's partitions, using the same consumer group rebalancing protocol covered in Module 03.

KStream and KTable — Two Ways to Interpret the Same Kind of Data

Kafka Streams gives you two core abstractions for working with a topic, and choosing the right one for a given topic is the first design decision in any Kafka Streams application.

KStream — a record stream, every event is independent

A KStream treats every record on a topic as an independent, immutable event. A KStream of order events is a sequence of "order placed" facts — each one meaningful on its own, none of them superseding or replacing another. Reading a KStream never "loses" an earlier record in favor of a later one with the same key; every record is retained and processed.

KTable — a changelog, the latest value per key is all that matters

A KTable treats a topic as a changelog of updates to a keyed table: each record represents the current, latest state for its key, and a new record with the same keyreplaces the previous value rather than adding to a sequence. A KTable of customer profiles keyed by customer_id represents "what is the current state of each customer" — conceptually the same relationship a compacted topic (Module on log compaction) has to its keys, and in fact KTables are commonly backed by, or materialized as, compacted topics internally for exactly this reason.

Records arriving, in order, key = customer_id:

  key=C1  value={"tier": "silver"}
  key=C2  value={"tier": "gold"}
  key=C1  value={"tier": "gold"}     <- C1 upgraded

As a KStream (customerTierChanges):
  Three independent events are seen and can each be processed --
  e.g. "send a congratulations email on every tier upgrade event"
  needs to see the transition, not just the final state.

As a KTable (customerCurrentTier):
  The table's current state, after all three records:
    C1 -> gold   (the second C1 record replaced the first)
    C2 -> gold
  A join against this KTable always sees the LATEST tier for a
  customer, never an intermediate value that has since changed.

Kafka Streams also has a third, less commonly used abstraction, GlobalKTable, which is a KTable fully replicated to every application instance rather than partitioned across them — useful for small reference datasets (a country-code lookup table, a small product catalog) that every instance needs complete local access to for joins, without caring which partition a given key would normally land on.

Stateless Operations Need No Memory. Stateful Operations Need Local Storage.

Every operation in Kafka Streams falls into one of two categories, and the distinction determines whether Kafka Streams needs to maintain any durable local state on your behalf.

Stateless operations — process each record independently

A stateless operation transforms or filters each record using only that record's own contents — it never needs to remember anything about records it has already seen. map (transform each record), filter (keep or drop each record based on a predicate), andbranch (split a stream into multiple streams based on a predicate) are the canonical examples. These operations are cheap: no local storage, no changelog topic, no state to rebuild after a restart.

Stateful operations — need to remember something across records

A stateful operation needs information beyond the current record to compute its result — aggregations (a running count, sum, or average per key) and joins (matching a record against previously-seen records from another stream or table) are the two major categories. Since Kafka Streams applications can be restarted, rebalanced, or scaled, this "memory" cannot simply live in a plain in-memory variable — it needs to be durable and recoverable. Kafka Streams solves this with local state stores, typically backed by RocksDB (an embedded key-value store that lives on local disk, extremely fast for the small, localized reads and writes a state store needs), one instance per application instance, holding only the portion of state relevant to the partitions that instance currently owns.

Every stateful operation's state store is backed by a changelog
topic -- an internal, compacted Kafka topic that Kafka Streams
creates and manages automatically, one per state store.

Every update to the local RocksDB state store is ALSO written to
its changelog topic, roughly like a write-ahead log:

  local state store: user_txn_count[user_42] = 7
       |
       v  (also written to)
  changelog topic:  key=user_42  value=7

If the application instance crashes and its work is reassigned to
a different instance (or the same instance restarts on a fresh
disk), the new owner does NOT recompute from the entire history of
the input topic. It instead replays the much smaller, compacted
changelog topic to rebuild just the latest state per key into a
fresh local RocksDB store -- fast recovery, because the changelog
only has to replay the latest value per key, not every input event
ever processed.

Windowing — Bounding an Otherwise-Unbounded Aggregation by Time

"Count all transactions for this user" is not a well-defined question on an unbounded stream — the count would simply grow forever, with no moment at which it is "done" or meaningfully comparable across users. "Count transactions for this user in the last 5 minutes" is well-defined, because it bounds the otherwise-infinite aggregation to a specific slice of time. This is what windowing is for: grouping stream-time into fixed-size buckets so aggregations produce a meaningful, finite result per bucket instead of one number that only ever grows.

Tumbling windows — fixed-size, non-overlapping

A tumbling window divides time into fixed-size, back-to-back, non-overlapping buckets. A 5-minute tumbling window produces buckets of [00:00-00:05), [00:05-00:10), [00:10-00:15), and so on — every event belongs to exactly one window, with no overlap and no gaps.

Hopping windows — fixed-size, overlapping

A hopping window is also fixed-size, but advances ("hops") by an interval smaller than the window size, so windows overlap and a single event can belong to more than one window. A 10-minute hopping window that advances every 5 minutes produces buckets [00:00-00:10), [00:05-00:15), [00:10-00:20) — each event falls into two overlapping windows, useful for smoother, more frequently-updated aggregates (a rolling 10-minute average recomputed every 5 minutes, rather than jumping discretely every 10 minutes).

Sliding windows — driven by event pairs, not a fixed clock

A sliding window (specifically as Kafka Streams defines it) is defined relative to pairs of events rather than fixed clock boundaries — two events fall in the same sliding window if they occur within the window's time difference of each other. This is most useful for join-like "events near each other in time" questions rather than fixed-interval reporting buckets.

Tumbling window, size=5min:
  |--- W1: 0-5 ---|--- W2: 5-10 ---|--- W3: 10-15 ---|
  Every event belongs to exactly ONE window. No overlap.

Hopping window, size=10min, advance=5min:
  |------ W1: 0-10 ------|
        |------ W2: 5-15 ------|
              |------ W3: 10-20 ------|
  An event at t=7 belongs to BOTH W1 and W2.
  Windows overlap; each event can update multiple window results.

Event-time vs processing-time — why the distinction changes your answer

Every windowing decision depends on which clock is used to place an event into a window. Event time is the timestamp of when the event actually happened in the real world, usually carried as a field in the record itself or as the Kafka record's embedded timestamp set by the producer. Processing time is the timestamp of when the stream processing application happens to handle the record, which can lag behind event time by anywhere from milliseconds to hours, depending on network delays, producer retries, or a consumer catching up after downtime.

Using processing time for a "transactions per 5 minutes" fraud aggregation means a burst of delayed events arriving all at once — say, after a network partition resolves — gets counted in whatever window happens to be open when they finally arrive, not the window they actually occurred in. This can produce a materially wrong answer to "how many transactions did this user make between 2:00 and 2:05," which is why Kafka Streams defaults to event-time semantics for windowing, extracting a timestamp from each record via a configurable timestamp extractor, rather than defaulting to the wall-clock time of the processing machine.

Joining Streams and Tables — Enriching Events With Context

A join combines records from two sources that share a common key, and Kafka Streams supports several shapes of join, each with different semantics driven by whether each side is a KStream (a sequence of events) or a KTable (a current-state snapshot).

KStream-KTable join — enrich each event with the latest known state

A KStream-KTable join is the most common enrichment pattern: for every incoming event on the KStream side, look up the current value for that event's key in the KTable, and combine them. This join is not windowed and does not wait — it uses whatever the KTable's latest value happens to be at the moment the KStream event is processed. Enriching an order event with the customer's current tier (from a KTable of customer profiles) is the canonical example: every order gets joined against whatever the customer's tier currently is, not the tier at some point in the past.

KStream-KStream join — needs a join window, because two streams never "finish"

A KStream-KStream join matches events from two unbounded streams that share a key and occurred near each other in time — for example, joining a stream of "page view" events with a stream of "add to cart" events, keyed by session ID, to find sessions where a view was quickly followed by a cart addition. Because neither stream ever "finishes," Kafka Streams cannot wait indefinitely hoping a matching event on the other side eventually arrives — doing so would mean holding every unmatched event in memory forever. This is why a KStream-KStream join always requires an explicit join window: only events on both sides that fall within that time window of each other are considered a match.

KTable-KTable join — a join of two current-state views

A KTable-KTable join combines two changelogs on a shared key into a new KTable representing the combined current state — for example, joining a KTable of customer profiles with a KTable of customer loyalty-program status to produce one combined "current customer view" KTable. Like the KStream-KTable join, this is not windowed — it always reflects the latest state on both sides.

Exactly-Once Processing in Kafka Streams — Built on Kafka Transactions

A Kafka Streams application, at its core, is a read-process-write loop: consume input records, update local state, produce output records. By default this is at-least-once — a crash between updating local state and having that update durably reflected can cause the same input to be reprocessed on recovery, potentially producing duplicate output records or double-counting an aggregation.

Setting processing.guarantee=exactly_once_v2 changes this. Under the hood, Kafka Streams uses exactly the transactional producer mechanism covered in the message-brokers module: every batch of output records produced, every state store update reflected in its changelog topic, and the consumer offset commit for the corresponding input records are wrapped in a single Kafka transaction. Either the entire unit — outputs, state changes, and input offset advancement — commits atomically, or none of it does. On a crash mid-transaction, the transaction is aborted, and any partial output is invisible to downstream consumers configured withisolation.level=read_committed.

Without exactly_once_v2 (at_least_once, the default):
  1. consume input record
  2. update local state store (+ write to changelog topic)
  3. produce output record
  4. commit input offset
  -- if the process crashes between steps 2 and 4, on restart the
     same input record is reprocessed: state may be double-updated,
     and a duplicate output record may be produced.

With exactly_once_v2:
  BEGIN TRANSACTION
    update local state store (+ changelog write)
    produce output record
    commit input offset  (as part of the SAME transaction)
  COMMIT TRANSACTION
  -- a crash before commit means the whole transaction is aborted;
     on restart, the input record is reprocessed as if nothing had
     happened yet -- no partial state update, no duplicate output,
     because nothing partial was ever visible to begin with.

Interactive Queries — Reading a State Store Without a Separate Database

A Kafka Streams application's state stores hold genuinely useful, continuously up-to-date data — the running transaction count per user from Part 08's worked example, or a materialized KTable of current customer tiers. The naive way to expose that to the rest of your organization is to also write it out to an external database and query that database instead. Interactive Queries offer a different option: querying a Kafka Streams application's own local state stores directly, over a thin API you build yourself (commonly a small REST endpoint inside the same application), without standing up or synchronizing a separate datastore at all.

The complication is that a state store's data is partitioned across every instance of the application — a single instance only holds the state for the partitions it currently owns, not the whole picture. Kafka Streams exposes metadata about which application instance owns which key, so a query received by any instance can either answer it locally (if it owns the relevant partition) or forward the request to the instance that does.

ReadOnlyWindowStore<String, Long> store = streams.store(
    StoreQueryParameters.fromNameAndType(
        "user-txn-counts-5min",
        QueryableStoreTypes.windowStore()
    )
);

// Does THIS instance own the partition for "U-88213"?
KeyQueryMetadata meta = streams.queryMetadataForKey(
    "user-txn-counts-5min", "U-88213", keySerde.serializer()
);

if (meta.activeHost().equals(thisInstance)) {
    // Answer locally -- this instance owns the relevant partition
    WindowStoreIterator<Long> results = store.fetch(
        "U-88213", Instant.now().minus(Duration.ofMinutes(5)), Instant.now()
    );
} else {
    // Forward the HTTP request to meta.activeHost() instead --
    // that instance is the one that actually owns this key's data
}

The Processor API — For When the DSL Isn't Expressive Enough, and Testing Without a Real Cluster

Everything covered so far — map, filter, groupByKey,windowedBy, joins — is part of the Kafka Streams DSL (domain-specific language), a high-level, declarative API that covers the large majority of real stream processing needs. Underneath the DSL sits the lower-level Processor API, which gives direct control over the processing topology: custom logic per record, direct access to state stores, and the ability to schedule periodic work independent of record arrival (via punctuate), none of which the DSL exposes directly.

Most Kafka Streams applications never need the Processor API — the DSL's built-in operations, including the ability to drop down into a custom transform or processstep within an otherwise-DSL topology, cover nearly everything. Reach for the full Processor API when you need genuinely custom control flow, such as emitting a result on a fixed wall-clock schedule regardless of whether new records have arrived, which the record-driven DSL has no clean way to express.

Testing a topology without a running Kafka cluster

A meaningful advantage of Kafka Streams being a library rather than a submitted job to an external engine is that its topology can be tested with an in-memory driver, TopologyTestDriver, with no real Kafka broker, Zookeeper, or KRaft controller running at all — a genuine unit test, not an integration test requiring test infrastructure.

TopologyTestDriver testDriver = new TopologyTestDriver(builder.build(), props);

TestInputTopic<String, Transaction> input = testDriver.createInputTopic(
    "freshcart.transactions", Serdes.String().serializer(), transactionSerde.serializer()
);
TestOutputTopic<String, FraudAlert> output = testDriver.createOutputTopic(
    "freshcart.fraud-alerts", Serdes.String().deserializer(), fraudAlertSerde.deserializer()
);

// Push 9 transactions for the same user within one window
for (int i = 0; i < 9; i++) {
    input.pipeInput("U-1", new Transaction("U-1", 42.00), baseTime.plusSeconds(i * 10));
}

// Assert an alert was produced -- 9 > threshold of 8
assertFalse(output.isEmpty());
FraudAlert alert = output.readValue();
assertEquals(9, alert.getTransactionCount());

Scaling a Kafka Streams Application — Tasks, Standby Replicas, and the Partition Ceiling

Kafka Streams divides a topology's work into stream tasks, where the number of tasks is determined by the number of partitions on the input topics — the same partition-driven parallelism ceiling covered for consumer groups in Module 03 applies directly here, because a Kafka Streams application's underlying mechanism is still ordinary consumer group partition assignment. A topology reading from an 8-partition topic has, at most, 8 stream tasks, and running more than 8 application instances means some instances sit idle with no tasks assigned, exactly as an oversized consumer group does.

freshcart.transactions has 8 partitions.

2 application instances running:
  instance-1: tasks for partitions [0,1,2,3]
  instance-2: tasks for partitions [4,5,6,7]

Scale to 4 application instances:
  instance-1: tasks for partitions [0,1]
  instance-2: tasks for partitions [2,3]
  instance-3: tasks for partitions [4,5]
  instance-4: tasks for partitions [6,7]

Scale to 10 application instances (beyond the partition count):
  8 instances get exactly one partition's tasks each.
  2 instances get NO tasks at all -- idle, doing nothing, exactly
  as an oversized consumer group behaves.

Standby replicas — trading extra resource cost for faster recovery

Part 04 covered that a state store's changelog topic is what allows state to be rebuilt after a crash or rebalance — but replaying a changelog topic from scratch still takes time proportional to how much state exists, during which queries against that store (including Interactive Queries from Part 08) return incomplete results. Setting num.standby.replicas to 1 or more tells Kafka Streams to maintain additional, continuously-updated replicas of each state store on other application instances — essentially a hot standby, kept current by consuming the same changelog topic in real time rather than only reading it during recovery. When a rebalance moves a task, if a standby replica for that task's state already exists and is current on the instance it's assigned to, the instance can serve from it almost immediately instead of replaying the changelog from scratch.

Repartitioning — When Kafka Streams Silently Creates Its Own Internal Topics

Every stateful operation in Kafka Streams requires its input to be correctly partitioned by the key the operation groups on — the same key must always land on the same partition, or an aggregation or join could see only part of a given key's data on any one application instance. Whenever you re-key a stream with groupBy, map, or aselectKey call that changes the record's key, Kafka Streams cannot simply continue processing in place — it has to route each record to whichever partition the new key belongs on, which may be a different partition, and possibly owned by a different application instance entirely.

Kafka Streams handles this automatically by creating an internal repartition topic: it produces the re-keyed records to this new, internally-managed topic, and then consumes from it as if it were the actual input to the next step — effectively performing a full produce-and-consume round trip through Kafka in the middle of the topology, invisible in the DSL code but very real in terms of latency and Kafka throughput consumed.

KStream<String, Transaction> transactions = builder.stream("freshcart.transactions");
// Assume this topic is keyed by transaction_id, NOT user_id

// groupByKey() -- uses the EXISTING key (transaction_id). No repartition,
// but also not useful here since we want to aggregate per USER.
transactions.groupByKey().count();

// groupBy() -- re-keys by user_id, which differs from the topic's
// existing key. Kafka Streams MUST create an internal repartition
// topic to route records to the correct partition for their new key.
transactions
    .groupBy((txnId, txn) -> txn.getUserId())   // <- re-keying happens here
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofSeconds(30)))
    .count();
// Internally creates: fraud-txn-count-detector-<generated-name>-repartition

Monitoring — What to Watch on a Running Kafka Streams Application

Because a Kafka Streams application is your own deployed service, it inherits whatever application-level monitoring your organization already runs — CPU, memory, restart counts — but it also exposes a specific set of Kafka Streams and underlying consumer/producer metrics that matter for diagnosing stream-processing-specific problems no generic application metric would catch.

Most of these metrics are exposed the standard way any JVM application exposes metrics — JMX, with a Prometheus JMX exporter sidecar being the most common production pattern — so they flow into the same dashboards and alerting infrastructure the rest of an organization's services already use. There is no Kafka-Streams-specific monitoring stack to stand up; the only real work is knowing which of these particular metrics matter for a stream processing workload specifically, since a generic "CPU and memory look fine" dashboard would miss every one of the problems this table actually catches.

The single habit worth building early: treat consumer-group-style lag and rebalance metrics, state store restoration progress, and record lateness as a package deal for any stateful, windowed application, the same way Part 10 (Error Library) and Part 09b (scaling) both assume you already have visibility into all three before diagnosing a live incident — reconstructing that visibility for the first time during an active fraud-detection outage is a much worse position to be in.

Deploying a Kafka Streams Application — Ordinary Service Infrastructure

Because Kafka Streams is a library rather than a submitted job, deploying it is genuinely no different from deploying any other JVM service — a JAR or container image, built through the same CI/CD pipeline as everything else, running under Kubernetes, ECS, or whatever your platform standardizes on. There is no Kafka-Streams-specific deployment artifact type and no separate scheduler to submit a job to.

# Dockerfile
FROM eclipse-temurin:17-jre
COPY target/fraud-txn-count-detector.jar /app/app.jar
ENTRYPOINT ["java", "-jar", "/app/app.jar"]

# Kubernetes deployment -- an ordinary Deployment, nothing
# Kafka-Streams-specific about the manifest itself
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fraud-txn-count-detector
spec:
  replicas: 4    # <- scaling is literally this number, per Part 09b
  template:
    spec:
      containers:
        - name: app
          image: freshcart/fraud-txn-count-detector:1.4.0
          env:
            - name: KAFKA_BOOTSTRAP_SERVERS
              value: "broker-1:9092,broker-2:9092"
          volumeMounts:
            - name: state-dir
              mountPath: /var/kafka-streams   # local RocksDB state, Part 04
      volumes:
        - name: state-dir
          persistentVolumeClaim:
            claimName: fraud-detector-state

The one Kafka-Streams-specific deployment consideration worth calling out is the local state directory (state.dir) — since state stores are RocksDB files on local disk, backed by changelog topics for recoverability per Part 04, mounting a persistent volume rather than ephemeral container storage means a pod restart on the same node can reuse its existing local state instead of always replaying the full changelog from scratch, shortening recovery time without needing standby replicas at all for that specific failure mode.

A Worked Example — Flagging Users Over a Transaction-Count Threshold in a 5-Minute Window

FreshCart's fraud team wants a real-time signal: flag any user who makes more than 8 card transactions within any 5-minute tumbling window, a pattern strongly associated with card-testing fraud (an attacker rapidly trying small transactions on a stolen card number to check whether it is still valid). This is a textbook stateful, windowed aggregation — the exact shape of problem Kafka Streams' DSL is built for.

Step 1 — model the input as a KStream, keyed by user

The source topic, freshcart.transactions, carries one record per transaction. Each record's key is already the user_id, which matters because Kafka Streams'groupByKey avoids an unnecessary repartition when the stream is already keyed correctly for the aggregation that follows — re-keying with groupBy instead would force a repartition topic and added latency for no reason here.

Step 2 — group by key, window, and count

This is expressed with the Kafka Streams DSL as a chain: read the topic as a KStream, group the already-keyed stream, apply a 5-minute tumbling window, and count records per user per window.

StreamsBuilder builder = new StreamsBuilder();

KStream<String, Transaction> transactions = builder.stream(
    "freshcart.transactions",
    Consumed.with(Serdes.String(), transactionSerde)
);

KTable<Windowed<String>, Long> txnCountsPerWindow = transactions
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(
        Duration.ofMinutes(5),
        Duration.ofSeconds(30)   // grace period for late-arriving events
    ))
    .count(Materialized.as("user-txn-counts-5min"));

KStream<String, FraudAlert> alerts = txnCountsPerWindow
    .toStream()
    .filter((windowedUserId, count) -> count > 8)
    .map((windowedUserId, count) -> KeyValue.pair(
        windowedUserId.key(),
        new FraudAlert(
            windowedUserId.key(),
            count,
            windowedUserId.window().startTime(),
            windowedUserId.window().endTime()
        )
    ));

alerts.to("freshcart.fraud-alerts", Produced.with(Serdes.String(), fraudAlertSerde));

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "fraud-txn-count-detector");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker-1:9092,broker-2:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Walking through what this actually does at runtime: every transaction record for a given user updates that user's count in the current 5-minute window's local state store (RocksDB-backed, per Part 04, with the update also written to the operation's changelog topic for fault tolerance).windowedBy means the count is tracked separately per window — a user's count in window [12:00-12:05) is entirely independent of their count in [12:05-12:10). The downstreamfilter(count > 8) only lets through window results that actually exceed the threshold, and the result is produced to a dedicated freshcart.fraud-alerts topic that a separate alerting service consumes.

Step 3 — what a downstream alert actually looks like

{
  "user_id": "U-88213",
  "transaction_count": 11,
  "window_start": "2026-09-11T14:35:00Z",
  "window_end": "2026-09-11T14:40:00Z"
}

Alert consumed by fraud-response-service:
  User U-88213 made 11 transactions between 14:35 and 14:40 UTC
  -> threshold (8) exceeded
  -> triggers: temporary card hold + risk-team review queue
  -> latency from 8th transaction to alert produced: ~340ms

Why event-time and the grace period both matter here

If this topology used processing time instead of event time, a burst of delayed transaction events arriving together after a brief network hiccup upstream could all land in whatever window happens to be open at the moment they finally arrive — potentially causing a false fraud alert for a user whose transactions were actually spread across a much longer real-world period, or missing a genuine pattern whose events get scattered into windows they did not actually belong to. The Duration.ofSeconds(30) grace period in the code above accepts transactions that arrive up to 30 seconds after their window's nominal end, still correctly attributed to the right window, before that window's result is finally considered closed.

Five Misconceptions About Stream Processing and Kafka Streams

What This Looks Like on Day One

5 Interview Questions — With Complete Answers

Mistakes Teams Make Building Kafka Streams Applications

Errors You Will Hit — And Exactly Why They Happen