order-events). Data is never modified in place — only appended. Retention is time- or size-based, not read-based.offset=0 is the oldest message. Consumers track their own offset per partition — Kafka itself does not track "what each consumer has read."replica.lag.time.max.ms.Producer Broker (leader, partition 0) Followers (ISR)
│ │ │
│── ProduceRequest ─────────────────►│ │
│ (batch of records, acks=all) │── Fetch (replicate) ─────────►│
│ │◄─ Ack (follower caught up) ───│
│ │── append to partition log │
│◄── ProduceResponse (offset=1042) ──│ │
│ (only sent after all ISR ack) │ │acks=all, the leader waits until all ISR replicas have written to their local log before acknowledging.Topic: order-events (4 partitions, replication-factor=3)
Broker 1: leader P0, follower P2, follower P3
Broker 2: leader P1, follower P0, follower P3
Broker 3: leader P2, follower P1, follower P0
Broker 4: leader P3, follower P1, follower P2
(spread leaders across brokers for even load)
Consumer Group "order-service" (3 consumers):
Consumer A → P0, P1 (one consumer can hold multiple partitions)
Consumer B → P2
Consumer C → P3
Consumer Group "audit-service" (2 consumers):
Consumer X → P0, P1, P2 (reads same data independently)
Consumer Y → P3
Consumer Group "analytics" (6 consumers):
Consumers 1–4 → P0–P3
Consumers 5–6 → IDLE ← adding consumers beyond partition count = wastereplica.lag.time.max.ms (default 30s). A lagging follower won't block producer acks.# Recommended production settings for durability:
acks=all
min.insync.replicas=2 # broker config per topic
replication.factor=3 # at topic creation
unclean.leader.election.enable=falseBefore KRaft, Kafka relied on ZooKeeper as an external metadata store: controller election, ISR changes, topic configs, and ACLs all lived in ZooKeeper znodes. Problems: operationally complex (two systems to maintain), controller re-election took 10–30 seconds on failure, practical partition limit ~200k.
KRaft (Kafka Raft, GA Kafka 3.3+) makes Kafka self-sufficient:
@metadata).| Semantic | Producer config | Consumer commit | Risk | When to use |
|---|---|---|---|---|
| At-most-once | acks=0 | Commit before processing | Message loss on crash | Metrics, logs where loss is tolerable |
| At-least-once | acks=all + retries | Commit after processing | Duplicates on retry | Most production systems + idempotent consumers |
| Exactly-once (EOS) | Idempotent + transactions | Atomic with produce | ~4% throughput overhead, complexity | Critical Kafka→Kafka pipelines (financial, ledger) |
The failure scenario for at-most-once is: producer sends, network drops the ack, producer does NOT retry → message lost. For most business events, losing data is worse than processing it twice. At-least-once accepts duplicates and pushes deduplication responsibility to the consumer — which you control.
EOS adds real operational weight: transactions need a coordinator, consumers must set isolation.level=read_committed, and any external system write (DB, REST API) falls outside the transaction boundary anyway.
props.put("enable.idempotence", "true");
// Automatically sets: acks=all, retries=MAX_INT, max.in.flight=5
// Kafka assigns each producer session a PID (Producer ID).
// Each message gets a sequence number per (PID, partition).
// Broker tracks: last accepted sequence per (PID, partition).
//
// On retry:
// producer resends with same PID + sequence number
// broker sees: "I already have sequence 42 for this PID+partition"
// broker deduplicates → no duplicate written to log
//
// Limitation: PID is per session. If producer restarts → new PID
// → sequence counter resets → broker can't deduplicate across restarts.props.put("transactional.id", "order-processor-1"); // stable across restarts
producer.initTransactions(); // registers with transaction coordinator broker
try {
producer.beginTransaction();
// Writes to multiple partitions/topics atomically:
producer.send(new ProducerRecord("debit-events", debitMsg));
producer.send(new ProducerRecord("credit-events", creditMsg));
// Atomically commits the consumer offset AND the produced messages.
// This is the read-process-write atomic unit:
producer.sendOffsetsToTransaction(currentOffsets, groupMetadata);
producer.commitTransaction(); // two-phase commit via coordinator
} catch (Exception e) {
producer.abortTransaction(); // marks records as aborted in log
}
// What happens under the hood:
// 1. beginTransaction: coordinator creates txn record in __transaction_state topic
// 2. Each produce: broker marks records as "transactional, pending"
// 3. commitTransaction: coordinator writes COMMIT marker to all involved partitions
// 4. Consumers with isolation.level=read_committed skip records until COMMIT markerprops.put("isolation.level", "read_committed");
// Consumer buffers records until the transaction COMMIT/ABORT marker arrives.
// Aborted records are discarded — never delivered to consumer.poll().
// Side effect: slightly higher end-to-end latency (waits for commit marker).
// Default is read_uncommitted — consumer sees all records including aborted ones.@Transactional
public void processOrder(OrderEvent event) {
// 1. Stable deduplication key — topic+partition+offset is naturally unique
String dedupKey = event.getEventId(); // or compose: topic+":"+partition+":"+offset
// 2. Check + process + mark — all in one DB transaction
if (processedEventRepo.existsById(dedupKey)) {
return; // already processed, skip
}
orderRepo.save(mapToOrder(event)); // business logic
processedEventRepo.save(new ProcessedEvent(dedupKey)); // mark done
// If crash here, DB transaction rolls back → both writes undone → safe retry
}The key: the DB lookup, business write, and deduplication marker write are all in one atomic transaction. Either all succeed or all roll back. No half-processed state.
A rebalance is the process where Kafka's Group Coordinator (a broker) redistributes partition ownership among consumers in a group. During a rebalance, all consumers in the group stop consuming until the new assignment is complete. This is called a "stop-the-world" rebalance.
Step 1: JoinGroup
All consumers send JoinGroup request to Group Coordinator broker.
Coordinator picks a "leader consumer" (usually the first one).
Step 2: SyncGroup
Leader consumer runs the partition assignment algorithm.
Sends assignments back to the Coordinator (SyncGroup request).
Coordinator distributes assignments to all other consumers.
Step 3: Consumers resume polling with their new partition assignment.
Timeline:
Consumer C crashes
↓ (session.timeout.ms = 45s elapses)
Coordinator detects timeout
↓
JoinGroup sent by remaining consumers
↓ (2–5 seconds for large groups)
Assignments redistributed
↓
Consumers resume — total pause ≈ 45s + 5s = ~50s worst case| Protocol | Behaviour | Downtime |
|---|---|---|
| Eager (old default) | All consumers revoke ALL partitions before reassignment. Then re-assign from scratch. | 100% of partitions paused during rebalance |
| Cooperative Sticky (Kafka 2.4+) | Only partitions that actually need to move are revoked. Others keep consuming throughout. | Only the moving partitions pause |
# Enable cooperative incremental rebalance (recommended):
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
# With CooperativeStickyAssignor:
# - Consumer A has P0, P1, P2
# - New consumer D joins
# - Only P2 is revoked from A and given to D
# - A keeps consuming P0 and P1 without interruption# Give each consumer instance a stable identity:
group.instance.id=order-consumer-pod-3 # e.g. Kubernetes pod name
session.timeout.ms=60000 # give restarts time to complete
# Without static membership:
# Pod restarts → consumer leaves group (triggers rebalance)
# → pod comes back → joins group (triggers another rebalance)
# = 2 rebalances per rolling restart, multiplied by pod count
# With static membership:
# Pod restarts → Coordinator holds the assignment for session.timeout.ms
# → Pod rejoins with same group.instance.id within timeout
# → Coordinator hands back the same partitions — NO rebalancegroup.instance.id to the pod name via the Downward API. Combined with CooperativeStickyAssignor, rolling deployments cause zero rebalances.session.timeout.ms = 45000 # broker-side: heartbeat timeout window
# consumer considered dead if no heartbeat received
# increased → slower failure detection
heartbeat.interval.ms = 3000 # how often background thread sends heartbeat
# must be < session.timeout.ms / 3
# does NOT affect max.poll.interval.ms
max.poll.interval.ms = 300000 # main thread: max time between poll() calls
# if exceeded, consumer voluntarily leaves group
# this catches "slow processing" cases
# even if heartbeat background thread is alive
max.poll.records = 500 # reduce to keep each poll() cycle fast
# fewer records = more frequent polls = healthier// WRONG: blocking the poll loop
while (true) {
var records = consumer.poll(Duration.ofMillis(100));
for (var record : records) {
callSlowExternalApi(record); // blocks for 30s → max.poll.interval exceeded → rebalance
}
}
// CORRECT: offload slow work, keep poll loop fast
while (true) {
var records = consumer.poll(Duration.ofMillis(100));
List<Future> futures = records.stream()
.map(r -> executor.submit(() -> callSlowExternalApi(r)))
.collect(toList());
futures.forEach(f -> f.get()); // wait for all async results
consumer.commitSync(); // then commit — all done
}The default partitioner computes: partition = murmur2(key) % numPartitions
// Producer sends with a key:
producer.send(new ProducerRecord<>("order-events", customerId, orderEvent));
// topic key value
// All events for customerId="user-123" → always partition 7 (for example)
// Consumer assigned to partition 7 sees all user-123 events in order| Key choice | Ordering guarantee | Trade-off |
|---|---|---|
customerId | All events per customer ordered | Good if customer event volume is even |
orderId | All events per order ordered | Fine-grained — many distinct keys, even distribution |
tenantId (mutable) | Per-tenant ordering | Danger: if tenantId changes (rename, M&A), same logical tenant maps to different partition — ordering breaks |
immutableTenantId (UUID) | Per-tenant ordering, stable | UUID is stable for the lifetime of data |
eventType | Per event type | Few distinct values → hot partitions |
| null | None | Even distribution, maximum parallelism |
// Problem: VIP tenant generates 80% of traffic
// All their events → partition 3 → Consumer C overwhelmed
// Other partitions idle
Topic "catalog-events" (8 partitions):
P0: 10 msg/s ← normal tenant
P1: 12 msg/s ← normal tenant
P2: 8 msg/s ← normal tenant
P3: 840 msg/s ← VIP tenant (hot!)
...
// Fix option 1: compound key with salt (breaks strict per-tenant ordering)
String key = tenantId + "-" + (eventSeq % 4);
// Spreads VIP tenant across 4 partitions.
// Trade-off: ordering is now per (tenantId, salt), not per tenantId.
// To reconstruct full order for that tenant → sort by event timestamp in consumer.
// Fix option 2: dedicated topic for high-volume tenants
// Route VIP tenants to "catalog-events-highvol" (more partitions)
// Normal tenants to "catalog-events" (fewer partitions)
// Fix option 3: pre-aggregate before publishing
// Batch multiple small events into one larger event per tenant → fewer messages// Problem: producer sends batch [A, B, C] to partition 3
// A succeeds, B fails (network blip), C succeeds
// Producer retries B → broker receives [A, C, B] → wrong order!
// This happens when max.in.flight.requests.per.connection > 1
// (multiple in-flight batches per partition)
// Fix: enable idempotent producer — safe with up to 5 in-flight:
enable.idempotence=true
// Idempotent producer uses sequence numbers per (PID, partition).
// Broker buffers out-of-order sequences and reorders them.
// Safe retries + correct ordering + no duplicates.
// Alternative (simpler but slower):
max.in.flight.requests.per.connection=1 // forces ordering but kills batching throughputIn a SaaS platform, creating a separate topic per tenant is impractical at scale. The pattern is to share topics but enforce isolation through layered controls:
// Layer 1: partition key = immutableTenantId (routing + ordering)
// Layer 2: envelope metadata (tenantId in message header or payload)
// Layer 3: consumer-side filter (skip messages not for this tenant)
// Layer 4: data-layer isolation (each tenant's data in separate DB schema)
// Layer 5: audit log per tenant (all events tagged)
// Consumer filter example:
records.stream()
.filter(r -> myTenantId.equals(r.headers().lastHeader("tenantId").value()))
.forEach(this::processEvent);This approach scales to thousands of tenants without partition explosion, while maintaining per-tenant ordering via the partition key.
An offset is a 64-bit integer that identifies a message's position within a single partition. It's monotonically increasing and never reused, even after a message is deleted by retention policy.
Offsets are per (consumer group, topic, partition) — stored in Kafka's internal topic __consumer_offsets. Two consumer groups reading the same topic have completely independent offsets.
Topic: order-events, Partition 3
offset=0: OrderCreated (Jan 1) ← may be deleted by retention
offset=1: OrderUpdated (Jan 2) ← may be deleted by retention
offset=2: OrderShipped (Jan 3)
...
offset=8429: OrderCancelled (today) ← log-end-offset
Consumer Group "fulfillment": committed offset = 8427
→ lag = 8429 - 8427 = 2 messages behindenable.auto.commit=true
auto.commit.interval.ms=5000 # commits every 5 seconds, regardless of processing state
// Timeline showing the problem:
//
// t=0: poll() returns records [100, 101, 102, 103, 104]
// t=2: auto-commit fires → commits offset=104 (all "read")
// t=3: consumer crashes mid-processing (processed 100, 101 only)
// t=4: consumer restarts → reads from committed offset=105
// → records 102, 103, 104 are SKIPPED FOREVER (data loss!)
//
// OR:
// t=0: poll() returns records [100..104]
// t=4.9: consumer finishes processing all 5 records
// t=5: auto-commit fires → commits 104
// t=5.1: commit fails (network error)
// t=5.2: consumer restarts → re-reads from 100 → DUPLICATESwhile (true) {
ConsumerRecords<String, Event> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, Event> record : records) {
processEvent(record.value()); // 1. process first
}
consumer.commitSync(); // 2. then commit — blocks until broker confirms
// commitSync() throws on failure → retry loop handles it
// if crash before commitSync() → restart re-reads same records (at-least-once, safe)
}
// commitAsync() vs commitSync():
// commitAsync() — non-blocking, ~2x throughput. But:
// - may fail silently (callback-only)
// - retrying async commits can cause out-of-order commits (later offset committed before earlier)
// Best practice: commitAsync() in the hot loop, commitSync() on shutdown/rebalance
// Hybrid pattern (high throughput + safe shutdown):
try {
while (true) {
var records = consumer.poll(Duration.ofMillis(100));
process(records);
consumer.commitAsync(); // fast path
}
} finally {
consumer.commitSync(); // guarantee last commit on shutdown
consumer.close();
}// For fine-grained control (e.g., processed records from multiple partitions):
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
for (ConsumerRecord<String, Event> record : records) {
processEvent(record);
// Commit after EACH record — lowest lag risk, but slow
offsets.put(
new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1) // +1 = "start reading from here next"
);
}
consumer.commitSync(offsets);
// Contiguous offset commit — only commit up to the last contiguous processed offset.
// Used in batch subscribers where some records may still be in-flight:
//
// Processed: [100, 101, _, 103, 104] (102 still processing)
// Safe to commit: offset=102 (not 104 — can't skip 102)// Consumer lag = log-end-offset − committed-offset (per partition)
// Represents: how many messages are waiting to be processed
// Monitor via:
// 1. JMX: kafka.consumer:records-lag-max
// 2. AdminClient API: compare group offsets vs log-end-offsets
// 3. Lag exporter tools: Burrow, kafka-consumer-lag-exporter → Prometheus Gauge
// Alert rules:
// lag > 10_000 AND processing_rate > 0 → consumer is slow, investigate
// lag > 10_000 AND processing_rate == 0 → consumer is dead, page on-call
// lag growing faster than message rate → consumer falling behind
// Production pitfall: lag is per partition, not per topic.
// A topic with 10 partitions may have lag=0 on 9 and lag=500_000 on 1 (hot partition).
// Always monitor per-partition lag, not just aggregate.auto.offset.reset=latest # (default) start reading from now — skip past messages
# used for: new consumer groups that don't need history
auto.offset.reset=earliest # start from beginning of the log (retention window)
# used for: catch-up consumers, replay scenarios
auto.offset.reset=none # throw exception if no offset found
# used for: critical consumers where "I don't know where I am" = bug
// Triggers: new consumer group, or committed offset has been deleted (past retention)
// Does NOT affect consumers with a valid committed offset.A service must atomically: (1) update the database and (2) publish a Kafka event. These are two different transactional systems. There is no XA/2PC that covers both. Any dual write has a failure window:
// Scenario A: crash after DB write, before Kafka send
db.save(order); // ✓ committed
// ← crash here
kafka.send(orderEvent); // never happens → downstream never knows
// Scenario B: Kafka send succeeds, DB rollback
kafka.send(orderEvent); // ✓ sent
db.save(order); // throws → rollback
// Downstream acted on an event for an order that doesn't exist in DB
// Both scenarios cause inconsistency. The Outbox pattern solves this.// outbox table schema:
CREATE TABLE outbox_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
topic VARCHAR NOT NULL, -- target Kafka topic
partition_key VARCHAR, -- used as Kafka message key
payload JSONB NOT NULL, -- the event data
created_at TIMESTAMP DEFAULT NOW(),
sent_at TIMESTAMP -- null until relayed
);
// Service writes both in the same DB transaction:
@Transactional
public void placeOrder(Order order) {
orderRepo.save(order); // business write
outboxRepo.save(OutboxEvent.of( // event write — same transaction
"order-events",
order.getId().toString(), // partition key
OrderEventMapper.toJson(order) // payload
));
// If anything fails → both writes roll back → consistent state
// DB commit succeeds → both writes persist → relay will find and publish the event
}| Approach | How it works | Pros | Cons |
|---|---|---|---|
| Debezium CDC | Reads DB transaction log (WAL/binlog). Captures every INSERT to outbox table in near-real-time. | Sub-second latency. No polling load on DB. No missed events. | Complex setup (Kafka Connect cluster). WAL retention must be configured. Schema changes can break connector. |
| Polling relay | Scheduled job: SELECT * FROM outbox WHERE sent_at IS NULL ORDER BY created_at LIMIT 100. Publishes, then marks sent. | Simple. No extra infrastructure. Works with any DB. | Polling interval = latency. DB load at scale. Thundering herd if backlog grows. Concurrent relay instances need locking. |
// Polling relay — safe concurrent version with optimistic locking:
UPDATE outbox_events
SET sent_at = NOW(), relay_id = :myInstanceId
WHERE id IN (
SELECT id FROM outbox_events
WHERE sent_at IS NULL
AND relay_id IS NULL -- not claimed yet
ORDER BY created_at
LIMIT 100
FOR UPDATE SKIP LOCKED -- skip rows locked by other relay instances
);
// Returns only the rows this instance claimed → no duplicate publishing// Debezium reads the DB transaction log (WAL for PostgreSQL, binlog for MySQL).
// It commits its WAL position to a Kafka Connect offset topic.
//
// On crash + restart:
// 1. Reads last committed WAL offset from Kafka Connect offset topic
// 2. Resumes reading WAL from that position
// 3. Re-emits any events that were published but not yet offset-committed
// → at-least-once delivery — your consumers must be idempotent
//
// Key config for PostgreSQL:
// wal_level=logical (in postgresql.conf)
// max_replication_slots=10 (each Debezium connector = 1 slot)
// max_wal_senders=10
//
// Debezium connector config (kafka-connect):
database.hostname=mydb.internal
database.port=5432
database.dbname=orders
table.include.list=public.outbox_events
transforms=outbox
transforms.outbox.type=io.debezium.transforms.outbox.EventRoutersent_at. On restart, it sees the row still unmarked → publishes again → duplicate event in Kafka. The Outbox pattern guarantees no event is lost, but does not guarantee no duplicates. Your consumers must be idempotent. Include a stable eventId (the outbox row UUID) in the event payload for consumers to use as a deduplication key.A poison pill is a message that always fails processing (malformed JSON, unexpected schema version, null field your code doesn't handle). The consumer retries → fails → retries → fails. Meanwhile:
// Partition consumption is blocked:
//
// Partition 3 message stream:
// offset=100: OrderCreated ← processed OK
// offset=101: ??? BAD JSON ← fails every time, stuck here
// offset=102: OrderUpdated ← CAN'T REACH — 101 must be committed first
// offset=103: OrderShipped ← CAN'T REACH
// ...
//
// Consumer keeps retrying 101. 102, 103, ... pile up.
// Partition lag grows. SLA violated. Downstream services wait.
// The entire partition is effectively dead until 101 is handled.// Spring Kafka DefaultErrorHandler with DLT — full config:
@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<?, ?> template) {
// Where to send failed messages:
DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
(record, ex) -> new TopicPartition(
record.topic() + ".DLT", // e.g. "order-events.DLT"
record.partition() // same partition — preserves ordering signal
));
// Retry policy — backoff before giving up:
ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
backOff.setInitialInterval(1000L); // 1s, 2s, 4s
backOff.setMultiplier(2.0);
backOff.setMaxInterval(10_000L); // cap at 10s
DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
// Don't waste retries on permanent failures — send to DLT immediately:
handler.addNotRetryableExceptions(
DeserializationException.class, // bad JSON/Avro — won't fix itself
DataIntegrityViolationException.class, // DB constraint — won't fix itself
IllegalArgumentException.class // validation error — won't fix itself
);
return handler;
}// Spring Kafka @RetryableTopic — creates retry topics automatically:
@RetryableTopic(
attempts = "4",
backoff = @Backoff(delay = 5000, multiplier = 6, maxDelay = 300_000),
dltStrategy = DltStrategy.FAIL_ON_ERROR,
topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE
)
@KafkaListener(topics = "order-events")
public void processOrder(OrderEvent event) { ... }
// Auto-created topics:
// order-events ← original
// order-events-retry-5000 ← retry after 5s
// order-events-retry-30000 ← retry after 30s
// order-events-retry-180000 ← retry after 3min
// order-events-dlt ← final dead letter
// Why this is better than simple retries:
// Main topic never slows down — poison pills are moved to retry topics.
// Retry topics have their own consumer groups with scheduled delays.
// If downstream service is down for 5 minutes, events wait in retry-3min — no partition blockage.// Spring Kafka automatically adds headers to DLT messages:
// kafka_dlt-original-topic = "order-events"
// kafka_dlt-original-partition = "3"
// kafka_dlt-original-offset = "101"
// kafka_dlt-exception-fqcn = "com.fasterxml.jackson.core.JsonParseException"
// kafka_dlt-exception-message = "Unexpected character..."
// kafka_dlt-original-timestamp = "1693000000000"
// Use these in your DLT consumer to:
// 1. Alert with context (which topic, which offset, what error)
// 2. Correlate with business data (use original-offset to find the raw message)
// 3. Decide replay strategy (can we fix and re-publish?)| Exception type | Retryable? | Examples | Action |
|---|---|---|---|
| Transient | Yes | ConnectException, TimeoutException, DB pool exhaustion | Retry with exponential backoff |
| Permanent | No | DeserializationException, DataIntegrityViolationException, NullPointerException on bad data | Skip retries, send directly to DLT |
| Unknown | Yes (conservative) | Any uncaught exception | Retry N times, then DLT |
The Consumer API is Kafka's low-level poll-based interface. You write the loop, manage offsets, handle errors, and build any stateful logic yourself.
// Consumer API is the right choice when:
// - Simple event forwarding: read from topic A, write to DB, call REST API
// - You need full control over offset commits (batch acknowledgement, per-record ack)
// - You're integrating with external systems (the "write to DB" part can't be inside a Kafka transaction)
// - Processing is stateless or state lives in an external DB
while (true) {
ConsumerRecords<String, Event> records = consumer.poll(Duration.ofMillis(100));
for (var record : records) {
myService.process(record.value()); // call DB, REST, etc.
}
consumer.commitSync();
}Kafka Streams is a client library (not a separate cluster) that builds on the Consumer/Producer APIs. It adds: a DSL for stream operations, automatic offset management, stateful operations backed by RocksDB, windowing, exactly-once for Kafka-to-Kafka, and fault-tolerant state recovery via changelog topics.
// Kafka Streams is the right choice when:
// - Aggregations: count events per key in 5-minute windows
// - Joins: enrich order events with customer data from another topic
// - Real-time analytics: running totals, sliding window fraud detection
// - The output is another Kafka topic (Kafka→Kafka pipelines)
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> orders = builder.stream("orders");
KTable<Windowed<String>, Long> orderCounts = orders
.filter((key, order) -> order.getAmount().compareTo(THRESHOLD) > 0)
.groupByKey()
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
.count(Materialized.as("order-count-store")); // backed by RocksDB
orderCounts
.toStream()
.filter((key, count) -> count > 10)
.to("fraud-alerts"); // output to Kafka topic// Every stateful operation (count, aggregate, join) uses a state store.
// Default implementation: RocksDB (embedded key-value store, persisted to disk).
//
// State store architecture:
//
// Kafka Streams app instance
// ├── Task 0 (owns partition 0)
// │ └── RocksDB state store (local disk)
// │ └── changelog topic: my-app-order-count-store-changelog-0
// ├── Task 1 (owns partition 1)
// │ └── RocksDB state store (local disk)
// │ └── changelog topic: my-app-order-count-store-changelog-1
// ...
//
// Every state change is written to a compacted changelog Kafka topic.
// On crash + restart: Streams replays the changelog to rebuild state.
// This can be slow for large state (millions of keys).
//
// Speed it up with standby replicas:
num.standby.replicas=1
// A second instance mirrors the state store continuously.
// On failure, it's promoted → near-instant recovery (state already warm).| Use case | Choose | Why |
|---|---|---|
| Read events, write to DB / REST API | Consumer API | External writes are outside Kafka transactions; simpler |
| Count/sum events per key over time windows | Kafka Streams | Built-in windowing + RocksDB state, fault-tolerant |
| Join two event streams (orders + payments) | Kafka Streams | KStream-KTable join with automatic co-partitioning |
| Simple topic-to-topic forwarding / fan-out | Consumer API | Overhead of Streams DSL not justified |
| Real-time fraud scoring (sliding windows) | Kafka Streams | Sliding window aggregation + low-latency read from state store |
| Complex event processing needing SQL | ksqlDB | SQL over Kafka Streams; consider if team prefers declarative |
// Topology: the DAG of processing nodes (source → processor → sink)
// Task: one unit of work = owns one input partition + its state stores
// Thread: a StreamThread runs multiple tasks
//
// Parallelism is controlled by:
// 1. Number of partitions in source topic (= max tasks)
// 2. num.stream.threads (tasks per instance)
// 3. Number of app instances (scale out)
//
// Example: 12 partitions, 3 instances, 2 threads each
// = 12 tasks, 6 threads total
// = each thread handles 2 tasks
num.stream.threads=4 // 4 StreamThreads per JVM instance
application.id=order-stats // unique ID = consumer group ID + changelog topic prefix
// NEVER change this in production without migrating state