In this post we are going to cover stream processing and Kafka Streams. These are 7 real Kafka interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: building fraud alerts that fire within seconds of a suspicious card payment.
Stream processing means we compute results continuously, as each event arrives, instead of waiting for a nightly job. Kafka Streams is the Java library that ships with Apache Kafka for this. Interviewers ask about it because live dashboards, late events, windows and joins come up on every streaming project, and the wrong choice gives wrong numbers.
Part 8 of the Real Time Kafka Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the cluster and some sample data
- Question 1: Your manager asks for live sales totals per minute. What is stream processing, and how is it different from batch?
- Question 2: Late events land in the wrong minute. What is the difference between event time and processing time?
- Question 3: A teammate asks which window to use for a dashboard. How do tumbling, hopping and session windows differ?
- Question 4: Someone confuses a KStream with a KTable. How do you explain the difference?
- Question 5: A stream-to-table join returns nothing for the first events. How do joins and state work in Kafka Streams?
- Question 6: Your manager asks whether to use Kafka Streams, Flink or Spark Structured Streaming. How do you decide?
- Scenario 7: The business wants fraud alerts within seconds of a suspicious card payment. How would you build it?
- The full script
The short answer
Stream processing updates results as each event arrives. Kafka Streams windows by event time (the record timestamp) with a grace period; a KStream is events, a KTable is the latest value per key; stream-table joins are lookups in local state. Pick Kafka Streams, Flink or Spark by sources, latency, state and the team.
Let's set up the cluster and some sample data
We use a local Kafka 4.4 cluster in KRaft mode with 1 broker for the first example. The other examples run in TopologyTestDriver, so they need no broker. Here are the starting files: Lab.java is a small test harness, then one Java class per question, and two Python scripts that send sales and read the totals.
import java.time.Instant;
import java.util.*;
import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.Windowed;
/** Test harness: runs a topology in TopologyTestDriver (no broker needed). */
public class Lab {
static final String DAY = "2026-10-26T";
static final Map<String, Long> windows = new TreeMap<>();
static TopologyTestDriver driver(StreamsBuilder b) {
Properties p = new Properties();
p.put(StreamsConfig.APPLICATION_ID_CONFIG, "lab");
p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "never-used:9092");
p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.StringSerde.class);
p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.StringSerde.class);
// lab only: no record cache, so every update is forwarded at once
// (the default is a 10 MB cache, flushed on commit)
p.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
p.put(StreamsConfig.STATE_DIR_CONFIG, "lab-state");
return new TopologyTestDriver(b.build(), p);
}
/** each record is "key=value@10:00:05" or "key@10:00:05" (value 1) */
static void pipe(TopologyTestDriver d, String topic, String... records) {
var ser = new StringSerializer();
var in = d.createInputTopic(topic, ser, ser);
for (String r : records) {
String[] kt = r.split("@"), kv = kt[0].split("=");
String value = kv.length > 1 ? kv[1] : "1";
System.out.println("in " + topic + " " + kt[0] + " at " + kt[1]);
in.pipeInput(kv[0], value, Instant.parse(DAY + kt[1] + "Z"));
}
}
static TopologyTestDriver run(StreamsBuilder b, String topic,
String... records) {
TopologyTestDriver d = driver(b);
pipe(d, topic, records);
return d;
}
static String hms(Instant t) { return t.toString().substring(11, 19); }
static String win(Windowed<String> w) {
return w.key() + " from " + hms(w.window().startTime())
+ " to " + hms(w.window().endTime());
}
/** keeps the latest count per window; null removes a merged session */
static void collect(Windowed<String> w, Long n) {
if (n == null) windows.remove(win(w)); else windows.put(win(w), n);
}
static void printWindows() {
windows.forEach((w, n) ->
System.out.println("window " + w + " count " + n));
}
static void dump(TopologyTestDriver d, String store) {
StringBuilder s = new StringBuilder(store + ":");
try (var it = d.getKeyValueStore(store).all()) {
it.forEachRemaining(kv -> s.append(" " + kv.key + "=" + kv.value));
}
System.out.println(s);
}
static void dropped(TopologyTestDriver d) {
d.metrics().forEach((name, m) -> {
if (name.name().equals("dropped-records-total"))
System.out.println("dropped late records: "
+ ((Number) m.metricValue()).intValue());
});
}
}import static java.time.Duration.*;
import java.util.*;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.kstream.Suppressed.BufferConfig;
public class SalesPerMinute {
public static void main(String[] args) {
var str = Serdes.String();
StreamsBuilder b = new StreamsBuilder();
b.stream("sales", Consumed.with(str, str))
.mapValues(amount -> Long.parseLong(amount))
.groupByKey()
.windowedBy(TimeWindows.ofSizeAndGrace(
ofMinutes(1), ofSeconds(10)))
.reduce(Long::sum, Materialized.with(str, Serdes.Long()))
.suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
.toStream((w, sum) -> w.key() + " " + w.window().startTime())
.mapValues(total -> total.toString())
.to("sales-per-minute", Produced.with(str, str));
var p = new Properties();
p.putAll(Map.of("application.id", "sales-per-minute",
"bootstrap.servers", "localhost:9092", "state.dir", "state"));
var app = new KafkaStreams(b.build(), p);
Runtime.getRuntime().addShutdownHook(new Thread(app::close));
app.start();
}
}import static java.time.Duration.*;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.kstream.Suppressed.BufferConfig;
public class LateEvents {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
b.<String, String>stream("swipes")
.groupByKey()
.windowedBy(TimeWindows.ofSizeAndGrace(
ofMinutes(1), ofSeconds(10)))
.count(Materialized.with(Serdes.String(), Serdes.Long()))
.suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
.toStream()
.foreach((w, n) -> System.out.println("out " + Lab.win(w)
+ " count " + n));
var d = Lab.run(b, "swipes", "card-7@10:00:10", "card-7@10:00:50",
"card-7@10:01:05", "card-7@10:00:55", "card-7@10:01:40",
"card-7@10:00:58", "card-7@10:02:30");
Lab.dropped(d);
d.close();
}
}import static java.time.Duration.*;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class Windows {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
KGroupedStream<String, String> clicks =
b.<String, String>stream("clicks").groupByKey();
var minute = TimeWindows.ofSizeWithNoGrace(ofMinutes(1));
var gap = SessionWindows.ofInactivityGapWithNoGrace(ofMinutes(1));
var counts = switch (args[0]) {
case "tumbling" -> clicks.windowedBy(minute).count();
case "hopping" ->
clicks.windowedBy(minute.advanceBy(ofSeconds(30))).count();
default -> clicks.windowedBy(gap).count();
};
counts.toStream().foreach(Lab::collect);
Lab.run(b, "clicks", "u1@10:00:10", "u1@10:00:40", "u1@10:01:20",
"u1@10:03:30", "u1@10:03:50").close();
Lab.printWindows();
}
}import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class StreamVsTable {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
KStream<String, String> balances = b.stream("balances");
// as a stream: every event counts
balances.groupByKey().count(Materialized.as("events-per-account"));
// as a table: only the latest value per key
balances.toTable(Materialized.as("latest-balance"));
var d = Lab.run(b, "balances", "acct-1=100@09:00:00",
"acct-2=50@09:05:00", "acct-1=80@09:10:00", "acct-1=120@09:20:00");
Lab.dump(d, "events-per-account");
Lab.dump(d, "latest-balance");
d.close();
}
}import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class Enrich {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
KTable<String, String> customers = b.table("customers");
b.<String, String>stream("payments")
.join(customers, (amount, tier) -> amount + " from a " + tier
+ " customer")
.foreach((id, v) -> System.out.println("out " + id + ": " + v));
var d = Lab.driver(b);
Lab.pipe(d, "payments", "c1=25@10:00:00"); // c1 not in the table
Lab.pipe(d, "customers", "c1=gold@10:00:05");
Lab.pipe(d, "payments", "c1=40@10:00:10");
d.close();
}
}import static java.time.Duration.*;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class FraudAlerts {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
b.<String, String>stream("card-payments")
.groupByKey()
.windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(
ofMinutes(1), ofSeconds(30)))
.count()
.toStream()
.filter((w, n) -> n != null && n == 3)
.foreach((w, n) -> System.out.println("ALERT " + Lab.win(w)
+ ": " + n + " payments"));
Lab.run(b, "card-payments", "card-9@12:00:05", "card-4@12:00:20",
"card-9@12:00:30", "card-9@12:00:50", "card-4@12:03:00",
"card-4@12:04:30").close();
}
}from datetime import datetime
from confluent_kafka import Producer
sales = [('store-1', 20, '10:00:05'), ('store-2', 15, '10:00:20'),
('store-1', 30, '10:00:45'), ('store-1', 25, '10:01:15'),
('store-2', 10, '10:01:30'), ('store-1', 5, '10:02:20')]
p = Producer({'bootstrap.servers': 'localhost:9092'})
for store, amount, t in sales:
event_time = datetime.fromisoformat('2026-10-26T%s+00:00' % t)
ms = int(event_time.timestamp() * 1000) # the event time
p.produce('sales', key=store, value=str(amount), timestamp=ms)
p.flush()
print('sent', len(sales), 'sales')from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'dashboard',
'auto.offset.reset': 'earliest'})
c.subscribe(['sales-per-minute'])
rows = []
while len(rows) < 4:
m = c.poll(1.0)
if m is not None and not m.error():
rows.append('%s total %s' % (m.key().decode(), m.value().decode()))
c.close()
print('\n'.join(sorted(rows)))First, we compile the Kafka Streams examples against the Kafka 4.4 jars with javac.
$ javac -proc:none -cp "libs/*" -d . *.javaQuestion 1: Your manager asks for live sales totals per minute. What is stream processing, and how is it different from batch?
Batch runs on a schedule over a finished set of data, and tells us what happened. Stream processing runs all the time, updates the result as each event arrives, and tells us what's happening now.
| Batch | Stream processing | |
|---|---|---|
| Input | a finished, bounded data set | an unbounded stream of events |
| Runs | on a schedule | all the time |
| Answers | what happened (minutes to hours later) | what is happening now (seconds) |
Let's look at the live totals as a Kafka Streams app, SalesPerMinute.java. It works in a clear sequence:
- Read the
salestopic. - Group by store.
- Use one-minute windows.
- Sum the amounts.
- Write one final total per store and minute to
sales-per-minute.
Kafka Streams is a library in our app, not a cluster. This is a plain Java program, so we start more copies with the same application.id to scale out, up to the input partition count. Its state lives in local stores, backed up to changelog topics.
Now let's run it on the real cluster, send six sales with sales.py, and read the totals with read_totals.py.
$ java -cp ".:libs/*" SalesPerMinute > app.log 2>&1 &
$ python sales.py
sent 6 sales
$ python read_totals.py
store-1 2026-10-26T10:00:00Z total 50
store-1 2026-10-26T10:01:00Z total 25
store-2 2026-10-26T10:00:00Z total 15
store-2 2026-10-26T10:01:00Z total 10Six sales came in, and four totals came out. store-1 sold 50 in the 10:00 minute and 25 in the next. The minute after that is still open, so there is no total for it yet.
Batch doesn't go away. The stream feeds the live dashboard, and a nightly batch reconciles the totals.
Question 2: Late events land in the wrong minute. What is the difference between event time and processing time?
Event time is when it happened, the record's timestamp. Processing time is when our app sees it. A card reader that was offline sends its swipes late, so by processing time they land in the wrong minute.
Kafka Streams windows by the record timestamp, so the producer must set it to the event time, or a TimestampExtractor reads it from the value. Its clock is stream time, the highest timestamp seen so far.
Let's run LateEvents.java: one-minute windows, a ten-second grace period, and suppress, to emit only final counts.
$ java -cp ".:libs/*" LateEvents
in swipes card-7 at 10:00:10
in swipes card-7 at 10:00:50
in swipes card-7 at 10:01:05
in swipes card-7 at 10:00:55
in swipes card-7 at 10:01:40
out card-7 from 10:00:00 to 10:01:00 count 3
in swipes card-7 at 10:00:58
in swipes card-7 at 10:02:30
out card-7 from 10:01:00 to 10:02:00 count 2
dropped late records: 1Let's walk through the output:
- The swipe from 10:00:55 arrives after a later one (10:01:05). It's out of order but within the grace, so it still counts in the 10:00 window.
- The final count, 3, comes out once stream time passes the window end plus grace.
- The swipe from 10:00:58 arrives after the window closed. Kafka Streams drops it, and the dropped records metric shows 1.
Pick the grace from how late your data really arrives. More grace means more complete, but later, results. Flink and Spark use watermarks for the same job.
Question 3: A teammate asks which window to use for a dashboard. How do tumbling, hopping and session windows differ?
Tumbling windows are fixed size and don't overlap. Hopping windows overlap. Session windows end after a gap of inactivity.
Let's run Windows.java three times on the same five clicks.
$ java -cp ".:libs/*" Windows tumbling
in clicks u1 at 10:00:10
in clicks u1 at 10:00:40
in clicks u1 at 10:01:20
in clicks u1 at 10:03:30
in clicks u1 at 10:03:50
window u1 from 10:00:00 to 10:01:00 count 2
window u1 from 10:01:00 to 10:02:00 count 1
window u1 from 10:03:00 to 10:04:00 count 2
$ java -cp ".:libs/*" Windows hopping
in clicks u1 at 10:00:10
in clicks u1 at 10:00:40
in clicks u1 at 10:01:20
in clicks u1 at 10:03:30
in clicks u1 at 10:03:50
window u1 from 09:59:30 to 10:00:30 count 1
window u1 from 10:00:00 to 10:01:00 count 2
window u1 from 10:00:30 to 10:01:30 count 2
window u1 from 10:01:00 to 10:02:00 count 1
window u1 from 10:03:00 to 10:04:00 count 2
window u1 from 10:03:30 to 10:04:30 count 2
$ java -cp ".:libs/*" Windows session
in clicks u1 at 10:00:10
in clicks u1 at 10:00:40
in clicks u1 at 10:01:20
in clicks u1 at 10:03:30
in clicks u1 at 10:03:50
window u1 from 10:00:10 to 10:01:20 count 3
window u1 from 10:03:30 to 10:03:50 count 2First, tumbling: each event is in exactly one window. Two clicks in the 10:00 minute, one in the next, two at 10:03. That's the classic per-minute chart.
Then hopping: windows are fixed size too, but advance by less than their size. One-minute windows every thirty seconds put each click in two windows, so we get six windows, from 09:59:30 to 10:04:30. Good for moving averages.
Next, session: no fixed size, a session ends after one minute of inactivity here. The first three clicks are one session (10:00:10 to 10:01:20, count 3), the last two another (count 2). Use them for user visits.
There are also sliding windows, which follow the events themselves, for rules like three payments within any minute.
Start with tumbling windows: each event counts once, and the totals add up. Hopping windows count each event more than once, so never sum them.
Question 4: Someone confuses a KStream with a KTable. How do you explain the difference?
A KStream is a stream of events: every record is a new fact, like a payment. A KTable is a changelog: each record is an update, and the table keeps the latest value per key, like an account balance.
| KStream | KTable | |
|---|---|---|
| A record is | a new event | an update |
| Example | payments | balances |
| Stored in | the topic | state store + changelog |
Let's read the same four records both ways with StreamVsTable.java.
$ java -cp ".:libs/*" StreamVsTable
in balances acct-1=100 at 09:00:00
in balances acct-2=50 at 09:05:00
in balances acct-1=80 at 09:10:00
in balances acct-1=120 at 09:20:00
events-per-account: acct-1=3 acct-2=1
latest-balance: acct-1=120 acct-2=50As a stream, every event counts: acct-1 has 3 events. As a table, there are two rows: acct-1 at 120, its latest balance, and acct-2 at 50.
Summing balances from a stream is the classic bug: it adds old balances on top of the new one. And counting payments from a table is wrong too, because each update replaces the last.
A KTable lives in a state store, backed by a compacted changelog topic, so it can be rebuilt after a crash. A GlobalKTable is a full copy of the table on every instance.
If a new record adds to the old one, it's a stream. If it replaces it, it's a table.
Question 5: A stream-to-table join returns nothing for the first events. How do joins and state work in Kafka Streams?
A stream-table join is a lookup. When a payment arrives, Kafka Streams looks up the customer in the table's local store, right then. It doesn't wait, and a table update alone produces no output.
Let's run Enrich.java, where the customer record arrives late.
$ java -cp ".:libs/*" Enrich
in payments c1=25 at 10:00:00
in customers c1=gold at 10:00:05
in payments c1=40 at 10:00:10
out c1: 40 from a gold customerThe first payment from c1 arrives before the customer record, so the inner join emits nothing. Then the customer arrives: still nothing. The next payment joins: "40 from a gold customer".
In production, the first events miss when table data arrives late, or has later timestamps, because Streams processes inputs in timestamp order. We have three fixes:
leftJoin: the event goes through without the details.- GlobalKTable: fully loaded before processing starts.
max.task.idle.ms: Streams waits for the table's data.
With a versioned table store, the join looks up the value as of the event's timestamp, so late events join the right version.
Both topics must be co-partitioned: same key, same partition count and the same partitioner (librdkafka's default is not Java's murmur2), unless we use a GlobalKTable. Stream-stream joins always need a JoinWindows.
Question 6: Your manager asks whether to use Kafka Streams, Flink or Spark Structured Streaming. How do you decide?
We decide on the sources, the latency, the state and the team. Let's look at each engine.
Kafka Streams is a Java library inside our service. No extra cluster to run, Kafka in and Kafka out, and exactly once with exactly_once_v2. Great for event-driven services owned by one team.
Flink is a distributed engine with its own cluster. It has the richest event time model, with watermarks, large state, many connectors, and Flink SQL. It's the pick for complex, low-latency pipelines across many sources.
Spark Structured Streaming runs micro-batches, and Spark 4.1 added a real-time mode. Pick it when the team already runs Spark or Databricks, and the output is lakehouse tables.
| Runs as | Sources and sinks | Time and latency | Best for | |
|---|---|---|---|---|
| Kafka Streams | a Java library in your app | Kafka in, Kafka out | stream time + grace; milliseconds | Kafka-centric services |
| Flink | its own cluster | many connectors, Flink SQL | watermarks; milliseconds | complex event time, big state |
| Spark Structured Streaming | a Spark cluster | many sources, lakehouse tables | watermarks; micro-batch, real-time mode (4.1+) | teams already on Spark |
So I decide on four things:
- Where the data comes from and goes to.
- The latency target.
- How much state and event time logic there is.
- What the team can already operate.
A per-minute total for one dashboard doesn't need a Flink cluster. A small Streams app is enough.
Scenario 7: The business wants fraud alerts within seconds of a suspicious card payment. How would you build it?
Let's build it step by step.
First, payments go to a card-payments topic keyed by card number, so one card's payments are in one partition, in order, on one stream task. That's what makes per-card state possible.
Then we start with simple rules in Kafka Streams. Three payments on one card within any minute raise an alert: a sliding window with thirty seconds of grace, a count, and a filter. Let's run FraudAlerts.java.
$ java -cp ".:libs/*" FraudAlerts
in card-payments card-9 at 12:00:05
in card-payments card-4 at 12:00:20
in card-payments card-9 at 12:00:30
in card-payments card-9 at 12:00:50
ALERT card-9 from 11:59:50 to 12:00:50: 3 payments
in card-payments card-4 at 12:03:00
in card-payments card-4 at 12:04:30card-9 pays three times in forty-five seconds, and the alert fires on the third payment. card-4 never gets to three. There is no suppress here: for fraud we want the result early. This test sets statestore.cache.max.bytes=0. With the default 10 MB cache, results are forwarded at the next commit: 30 s by default, 100 ms with exactly_once_v2, so alerts stay fast.
Next, we add context. We join each payment with a card-profiles KTable, like home country and usual amount, and score it with a model embedded in the app, not a remote call per payment. Alerts go to their own fraud-alerts topic with exactly_once_v2, and the alert service dedupes by alert ID.
- Latency: measure payment to alert, end to end.
- Rules in config: analysts tune the thresholds.
- Replay: run past payments through a new rule before switching it on.
Key points to remember
- Streaming updates results as events arrive; batch runs on a schedule
- Window by event time, with a grace period for late events
- Tumbling for dashboards, hopping overlaps, session by inactivity
- KStream = events, KTable = latest per key; joins are lookups
- Kafka Streams, Flink or Spark: sources, latency, state, team
The full script
This is a Python script that writes the files and runs every command in order against a local Kafka 4.4. Run pip install confluent-kafka first.
Show the full script that writes the files and runs every step
# ========================================================================
# Kafka Interview Q&A #8 - Stream processing and Kafka Streams
# TechBrothersIT. Sample data is invented for teaching.
# Every step below was run on Apache Kafka 4.4.0 (KRaft mode, 1 node) with confluent-kafka 2.16.0;
# the outputs in the video and the blog post are the real outputs. To run it yourself (Linux, macOS or WSL):
# 1. Download Kafka 4.4.0 from kafka.apache.org, extract it, and put its bin folder on your PATH.
# 2. Start a local cluster (see the Kafka quickstart; node 1 listens on localhost:9092).
# 3. pip install confluent-kafka then python Kafka_Interview_QA_08_Code.py in an empty folder.
# Steps that stop or start a broker say so in a comment: do that by hand at that point.
# ========================================================================
import os, subprocess
def write(path, text):
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True).returncode
if rc != 0 and not expect_error:
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: files used by the steps
write('Lab.java', """\
import java.time.Instant;
import java.util.*;
import org.apache.kafka.common.serialization.*;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.Windowed;
/** Test harness: runs a topology in TopologyTestDriver (no broker needed). */
public class Lab {
static final String DAY = "2026-10-26T";
static final Map<String, Long> windows = new TreeMap<>();
static TopologyTestDriver driver(StreamsBuilder b) {
Properties p = new Properties();
p.put(StreamsConfig.APPLICATION_ID_CONFIG, "lab");
p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "never-used:9092");
p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.StringSerde.class);
p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.StringSerde.class);
// lab only: no record cache, so every update is forwarded at once
// (the default is a 10 MB cache, flushed on commit)
p.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
p.put(StreamsConfig.STATE_DIR_CONFIG, "lab-state");
return new TopologyTestDriver(b.build(), p);
}
/** each record is "key=value@10:00:05" or "key@10:00:05" (value 1) */
static void pipe(TopologyTestDriver d, String topic, String... records) {
var ser = new StringSerializer();
var in = d.createInputTopic(topic, ser, ser);
for (String r : records) {
String[] kt = r.split("@"), kv = kt[0].split("=");
String value = kv.length > 1 ? kv[1] : "1";
System.out.println("in " + topic + " " + kt[0] + " at " + kt[1]);
in.pipeInput(kv[0], value, Instant.parse(DAY + kt[1] + "Z"));
}
}
static TopologyTestDriver run(StreamsBuilder b, String topic,
String... records) {
TopologyTestDriver d = driver(b);
pipe(d, topic, records);
return d;
}
static String hms(Instant t) { return t.toString().substring(11, 19); }
static String win(Windowed<String> w) {
return w.key() + " from " + hms(w.window().startTime())
+ " to " + hms(w.window().endTime());
}
/** keeps the latest count per window; null removes a merged session */
static void collect(Windowed<String> w, Long n) {
if (n == null) windows.remove(win(w)); else windows.put(win(w), n);
}
static void printWindows() {
windows.forEach((w, n) ->
System.out.println("window " + w + " count " + n));
}
static void dump(TopologyTestDriver d, String store) {
StringBuilder s = new StringBuilder(store + ":");
try (var it = d.getKeyValueStore(store).all()) {
it.forEachRemaining(kv -> s.append(" " + kv.key + "=" + kv.value));
}
System.out.println(s);
}
static void dropped(TopologyTestDriver d) {
d.metrics().forEach((name, m) -> {
if (name.name().equals("dropped-records-total"))
System.out.println("dropped late records: "
+ ((Number) m.metricValue()).intValue());
});
}
}
""")
write('SalesPerMinute.java', """\
import static java.time.Duration.*;
import java.util.*;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.kstream.Suppressed.BufferConfig;
public class SalesPerMinute {
public static void main(String[] args) {
var str = Serdes.String();
StreamsBuilder b = new StreamsBuilder();
b.stream("sales", Consumed.with(str, str))
.mapValues(amount -> Long.parseLong(amount))
.groupByKey()
.windowedBy(TimeWindows.ofSizeAndGrace(
ofMinutes(1), ofSeconds(10)))
.reduce(Long::sum, Materialized.with(str, Serdes.Long()))
.suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
.toStream((w, sum) -> w.key() + " " + w.window().startTime())
.mapValues(total -> total.toString())
.to("sales-per-minute", Produced.with(str, str));
var p = new Properties();
p.putAll(Map.of("application.id", "sales-per-minute",
"bootstrap.servers", "localhost:9092", "state.dir", "state"));
var app = new KafkaStreams(b.build(), p);
Runtime.getRuntime().addShutdownHook(new Thread(app::close));
app.start();
}
}
""")
write('LateEvents.java', """\
import static java.time.Duration.*;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.kstream.Suppressed.BufferConfig;
public class LateEvents {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
b.<String, String>stream("swipes")
.groupByKey()
.windowedBy(TimeWindows.ofSizeAndGrace(
ofMinutes(1), ofSeconds(10)))
.count(Materialized.with(Serdes.String(), Serdes.Long()))
.suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()))
.toStream()
.foreach((w, n) -> System.out.println("out " + Lab.win(w)
+ " count " + n));
var d = Lab.run(b, "swipes", "card-7@10:00:10", "card-7@10:00:50",
"card-7@10:01:05", "card-7@10:00:55", "card-7@10:01:40",
"card-7@10:00:58", "card-7@10:02:30");
Lab.dropped(d);
d.close();
}
}
""")
write('Windows.java', """\
import static java.time.Duration.*;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class Windows {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
KGroupedStream<String, String> clicks =
b.<String, String>stream("clicks").groupByKey();
var minute = TimeWindows.ofSizeWithNoGrace(ofMinutes(1));
var gap = SessionWindows.ofInactivityGapWithNoGrace(ofMinutes(1));
var counts = switch (args[0]) {
case "tumbling" -> clicks.windowedBy(minute).count();
case "hopping" ->
clicks.windowedBy(minute.advanceBy(ofSeconds(30))).count();
default -> clicks.windowedBy(gap).count();
};
counts.toStream().foreach(Lab::collect);
Lab.run(b, "clicks", "u1@10:00:10", "u1@10:00:40", "u1@10:01:20",
"u1@10:03:30", "u1@10:03:50").close();
Lab.printWindows();
}
}
""")
write('StreamVsTable.java', """\
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class StreamVsTable {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
KStream<String, String> balances = b.stream("balances");
// as a stream: every event counts
balances.groupByKey().count(Materialized.as("events-per-account"));
// as a table: only the latest value per key
balances.toTable(Materialized.as("latest-balance"));
var d = Lab.run(b, "balances", "acct-1=100@09:00:00",
"acct-2=50@09:05:00", "acct-1=80@09:10:00", "acct-1=120@09:20:00");
Lab.dump(d, "events-per-account");
Lab.dump(d, "latest-balance");
d.close();
}
}
""")
write('Enrich.java', """\
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class Enrich {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
KTable<String, String> customers = b.table("customers");
b.<String, String>stream("payments")
.join(customers, (amount, tier) -> amount + " from a " + tier
+ " customer")
.foreach((id, v) -> System.out.println("out " + id + ": " + v));
var d = Lab.driver(b);
Lab.pipe(d, "payments", "c1=25@10:00:00"); // c1 not in the table
Lab.pipe(d, "customers", "c1=gold@10:00:05");
Lab.pipe(d, "payments", "c1=40@10:00:10");
d.close();
}
}
""")
write('FraudAlerts.java', """\
import static java.time.Duration.*;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
public class FraudAlerts {
public static void main(String[] args) {
StreamsBuilder b = new StreamsBuilder();
b.<String, String>stream("card-payments")
.groupByKey()
.windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(
ofMinutes(1), ofSeconds(30)))
.count()
.toStream()
.filter((w, n) -> n != null && n == 3)
.foreach((w, n) -> System.out.println("ALERT " + Lab.win(w)
+ ": " + n + " payments"));
Lab.run(b, "card-payments", "card-9@12:00:05", "card-4@12:00:20",
"card-9@12:00:30", "card-9@12:00:50", "card-4@12:03:00",
"card-4@12:04:30").close();
}
}
""")
write('sales.py', """\
from datetime import datetime
from confluent_kafka import Producer
sales = [('store-1', 20, '10:00:05'), ('store-2', 15, '10:00:20'),
('store-1', 30, '10:00:45'), ('store-1', 25, '10:01:15'),
('store-2', 10, '10:01:30'), ('store-1', 5, '10:02:20')]
p = Producer({'bootstrap.servers': 'localhost:9092'})
for store, amount, t in sales:
event_time = datetime.fromisoformat('2026-10-26T%s+00:00' % t)
ms = int(event_time.timestamp() * 1000) # the event time
p.produce('sales', key=store, value=str(amount), timestamp=ms)
p.flush()
print('sent', len(sales), 'sales')
""")
write('read_totals.py', "from confluent_kafka import Consumer\n\nc = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'dashboard',\n 'auto.offset.reset': 'earliest'})\nc.subscribe(['sales-per-minute'])\nrows = []\nwhile len(rows) < 4:\n m = c.poll(1.0)\n if m is not None and not m.error():\n rows.append('%s total %s' % (m.key().decode(), m.value().decode()))\nc.close()\nprint('\\n'.join(sorted(rows)))\n")
run('ln -s "$(dirname "$(which kafka-topics.sh)")/../libs" libs')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic sales --partitions 1 --config retention.ms=-1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic sales-per-minute --partitions 1 --config retention.ms=-1')
# Step 1: Compile the Kafka Streams examples against the Kafka 4.4 jars
run('javac -proc:none -cp "libs/*" -d . *.java')
# Step 2: Run SalesPerMinute on the cluster, send six sales, read the totals per store and minute
run('timeout 300 java -cp ".:libs/*" SalesPerMinute > app.log 2>&1 & echo $! > app.pid')
run('python sales.py')
run('python read_totals.py')
# Step 3: Stop the Kafka Streams app (the shutdown hook closes it)
run('kill $(cat app.pid); while kill -0 $(cat app.pid) 2>/dev/null; do sleep 0.5; done')
# Step 4: Event time, grace period and a late event (TopologyTestDriver)
run('java -cp ".:libs/*" LateEvents')
# Step 5: Tumbling, hopping and session windows on the same five clicks (TopologyTestDriver)
run('java -cp ".:libs/*" Windows tumbling')
run('java -cp ".:libs/*" Windows hopping')
run('java -cp ".:libs/*" Windows session')
# Step 6: The same four records as a KStream and as a KTable (TopologyTestDriver)
run('java -cp ".:libs/*" StreamVsTable')
# Step 7: A stream-table join when the table record arrives late (TopologyTestDriver)
run('java -cp ".:libs/*" Enrich')
# Step 8: Fraud rule: three payments on one card within one minute (TopologyTestDriver)
run('java -cp ".:libs/*" FraudAlerts')
Part 9: Performance, Monitoring and Operations
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on Apache Kafka 4.4.0 in KRaft mode with confluent-kafka 2.16.0; topic IDs are masked. Part of the Real Time Kafka Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Apache Kafka documentation (kafka.apache.org) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.