Streams & Transactions

Solution and explanation

What it is

A KStream is Kafka Streams' abstraction of a **record stream**: an unbounded sequence of independent events. Each record is a new fact; a later record with the same key does **not** replace an earlier one.

KStream of payments (key = account):
ACC-1 → ₹500
ACC-2 → ₹200
ACC-1 → ₹300      both ACC-1 events stay; they are separate payments

Typical operations

KStream<String, Payment> payments = builder.stream("payments");

KStream<String, Payment> large = payments.filter((acc, p) -> p.amount() > 100_000);
KStream<String, String> alerts = large.mapValues(p -> "Large payment " + p.id());
alerts.to("payment-alerts");

// Branch by type
Map<String, KStream<String, Payment>> branches = payments.split(Named.as("pay-"))
    .branch((k, p) -> p.type() == UPI, Branched.as("upi"))
    .branch((k, p) -> p.type() == CARD, Branched.as("card"))
    .defaultBranch(Branched.as("other"));

• Stateless: filter, map, mapValues, flatMap, peek, merge, split.
• Re-keying: selectKey (triggers repartitioning before grouping).
• Stateful: groupByKey().count()/aggregate() turns a stream into a **KTable**.
• Joins: stream-stream (within a time window) and stream-table (enrichment).

Examples of things that are streams

Clicks, payments, application submissions, sensor readings, log lines: every event matters on its own.

Summary

A KStream is an append-only stream of independent events where every record counts; you filter, transform, join and aggregate it.

Solution and explanation

What it is

A KTable is Kafka Streams' abstraction of a **changelog stream**: it represents the **latest value per key**, like a database table continuously updated. A new record for a key is an **update** (upsert); a record with a null value is a **delete** (tombstone).

Topic customer-profiles (key = customerId):
C1 → {city: Pune}
C2 → {city: Delhi}
C1 → {city: Mumbai}     ← replaces C1's previous value

KTable state: C1 → Mumbai, C2 → Delhi

Creating and using

KTable<String, Customer> customers = builder.table("customer-profiles",
        Materialized.as("customers-store"));          // backed by a local state store

// Aggregations produce KTables
KTable<String, Long> ordersPerCustomer = orders.groupByKey().count();

// Enrich a stream with the current table value
KStream<String, EnrichedOrder> enriched = orders.join(customers,
        (order, customer) -> new EnrichedOrder(order, customer.city()));

The table is materialised in a **state store** (RocksDB) with a compacted changelog topic for recovery. Each instance holds only the partitions it owns.

GlobalKTable

A GlobalKTable replicates **all partitions to every instance**, so lookups need no co-partitioning; good for small reference data (currencies, branch codes, job categories).

Typical uses

Current customer profile, account balance, product price, inventory per SKU, latest job status: any "current state per key".

Summary

A KTable is a continuously updated table holding the latest value per key, built from a compacted topic or an aggregation, and used for lookups, joins and current-state views.

Solution and explanation

Core difference

• **KStream:** every record is an independent **event** (insert semantics). "Asha paid ₹500", "Asha paid ₹300" are two payments.
• **KTable:** every record is an **update** to the current value of its key (upsert semantics). "Asha's balance is ₹500", then "Asha's balance is ₹800": only ₹800 matters now.

Comparison

Duality

They convert into each other:

• **Stream → table:** aggregate a stream (groupByKey().count(), reduce), or treat the latest record per key as the value.
• **Table → stream:** table.toStream() emits every update as an event (a changelog).

KStream<String, Payment> payments = builder.stream("payments");
KTable<String, Long> totalPaid = payments
        .groupByKey()
        .aggregate(() -> 0L, (acc, p, total) -> total + p.amountPaise(),
                   Materialized.with(Serdes.String(), Serdes.Long()));
totalPaid.toStream().to("total-paid-per-account");

Joins

• stream-table join: enrich each event with the current state (order + customer profile).
• stream-stream join: match events within a time window (payment request + payment confirmation within 10 minutes).
• table-table join: maintain a combined view.

Summary

A KStream treats records as independent events, a KTable treats them as updates to the latest value per key; aggregations turn streams into tables and toStream() turns tables back into streams.

Solution and explanation

What it is

A state store is **local storage inside a Kafka Streams instance** used by stateful operations: counts, aggregations, windowed results, joins, deduplication and custom processors. By default it is **RocksDB** on local disk (in-memory stores are also available).

Fault tolerance through changelog topics

Every update to a store is also written to a compacted **changelog topic** in Kafka (<application.id>-<store>-changelog). If an instance dies, the task moves to another instance, which **restores the store by replaying the changelog**. Options:

• num.standby.replicas=1: keep warm copies on other instances for fast failover.
• Changelogs are compacted, so restoration reads only the latest value per key.

Using a store in a custom processor

StoreBuilder<KeyValueStore<String, Long>> seenStore = Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("seen-payment-ids"), Serdes.String(), Serdes.Long());
builder.addStateStore(seenStore);

payments.process(() -> new Processor<String, Payment, String, Payment>() {
    private KeyValueStore<String, Long> seen;
    private ProcessorContext<String, Payment> ctx;
    public void init(ProcessorContext<String, Payment> c) { ctx = c; seen = c.getStateStore("seen-payment-ids"); }
    public void process(Record<String, Payment> r) {
        if (seen.get(r.value().id()) != null) return;          // duplicate: drop
        seen.put(r.value().id(), r.timestamp());
        ctx.forward(r);
    }
}, "seen-payment-ids");

Interactive queries

Stores can be queried directly from the application, e.g. a REST endpoint returning the current count for a company, without a separate database. For keys on other instances, the app forwards the request using the streams metadata.

Sizing and operations

• Local disk must hold the state of the partitions the instance owns.
• Restoration time grows with state size; use standbys and compaction.
• Windowed stores need retention settings so old windows are removed.

Summary

State stores are local, RocksDB-backed key-value stores that power stateful stream processing, made fault-tolerant by compacted changelog topics and optional standby replicas.

Solution and explanation

What it is

A Kafka transaction lets a producer **write to multiple partitions and topics, and commit consumer offsets, as one atomic unit**: either everything becomes visible to read_committed consumers or nothing does. It is the foundation of exactly-once processing in Kafka.

Components

• **Transactional producer** with a unique transactional.id.
• **Transaction coordinator** (a broker) that records transaction state in the internal topic __transaction_state.
• **Commit/abort markers** written into each involved partition.
• **Consumers with isolation.level=read_committed**, which skip records from aborted transactions and wait for open ones.

Flow

props.put("transactional.id", "invoice-builder-" + instanceId);
KafkaProducer<String, Invoice> producer = new KafkaProducer<>(props);
producer.initTransactions();                       // registers, fences older instances

producer.beginTransaction();
try {
    producer.send(new ProducerRecord<>("invoices", orderId, invoice));
    producer.send(new ProducerRecord<>("ledger-entries", accountId, entry));
    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());   // input progress
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException e) {
    producer.close();                              // fatal: another instance owns this id
} catch (KafkaException e) {
    producer.abortTransaction();                   // nothing becomes visible
}

What it guarantees and what it does not

• Atomic writes across partitions plus offsets: yes.
• No zombie writes: an old instance with the same transactional.id is **fenced**.
• External side effects (database writes, emails, HTTP calls) are **not** part of the transaction.

Costs

Extra round trips to the coordinator and markers in the log; consumers using read_committed see data only after commit, adding latency equal to the transaction duration. Commit in batches (e.g. every 100 ms) rather than per record.

Summary

Kafka transactions make writes to several partitions plus consumer offset commits atomic, enabling exactly-once read-process-write pipelines inside Kafka.

Solution and explanation

What it is

transactional.id is a **stable, unique identifier for a transactional producer**. Setting it turns on transactions (and idempotence). The transaction coordinator maps it to a producer id and an **epoch**.

Why it must be stable

Its main job is **fencing zombies**. Suppose instance A is processing partition 0 and hangs (long GC pause). The group moves the work to instance B, which starts with the same transactional.id. When B calls initTransactions():

1. The coordinator **aborts** any open transaction of the previous holder.
2. It **increments the epoch**.
3. If A wakes up and tries to send or commit, its old epoch is rejected with ProducerFencedException.

So a frozen old instance can never write duplicate or conflicting output.

Choosing the value

• It must be the **same across restarts** of the same logical producer and **unique** among concurrently running producers.
• Classic approach: one id per input partition, e.g. orders-processor-<topic>-<partition>.
• Since Kafka 2.5 (KIP-447), fencing also uses consumer group metadata (sendOffsetsToTransaction(offsets, groupMetadata)), so one id per application instance (e.g. pod ordinal) is sufficient.

transactional.id=payment-processor-${HOSTNAME}     # StatefulSet: payment-processor-0, -1, -2

Spring Kafka: spring.kafka.producer.transaction-id-prefix=tx- generates ids per producer.

Expiry

If a transactional id is unused for transactional.id.expiration.ms (7 days), the coordinator forgets it.

Summary

transactional.id identifies a transactional producer across restarts so the coordinator can abort its unfinished transactions and fence older instances via epochs; keep it stable and unique.

Solution and explanation

What it does

isolation.level=read_committed makes a consumer **return only records from committed transactions** (plus non-transactional records). Records from aborted transactions are filtered out, and records from transactions still open are held back.

The default is read_uncommitted, which returns everything, including records that are later aborted.

isolation.level=read_committed

How it works

• The broker tracks the **Last Stable Offset (LSO)**: the offset before the earliest still-open transaction.
• A read_committed consumer can read only up to the LSO.
• Aborted records are skipped using the aborted-transaction index the broker returns with fetches.

Example

Partition: [A][B][T1:x][T1:y][C][T1 commit marker]
read_uncommitted: A B x y C
read_committed:   A B (waits until T1 commits) x y C
If T1 aborts:     A B C

Consequences

• **Correctness:** downstream services never act on output from failed transactions, essential for exactly-once pipelines (e.g. ledger entries).
• **Latency:** a long-running transaction blocks readers of that partition until it ends (lag grows). Keep transactions short and set transaction.timeout.ms sensibly (default 60 s).
• A hung producer with an open transaction stalls read_committed consumers until the timeout aborts it.

When to use it

Any consumer of topics written transactionally, including Kafka Streams with exactly-once (it sets this automatically). For topics with no transactional producers, it makes no difference.

Summary

read_committed hides aborted and in-progress transactional records by reading only up to the last stable offset, which is required for exactly-once but adds latency while transactions are open.

Solution and explanation

What it is

Log compaction is a retention mode (cleanup.policy=compact) where Kafka keeps **at least the latest record for every key** and removes older records for keys that have newer values. Instead of deleting by age, the topic becomes a durable, replayable **snapshot of the current state per key**.

Before: [K1:a][K2:b][K1:c][K3:d][K2:null][K1:e]
After:  [K3:d][K1:e]          (K2 deleted by tombstone after delete.retention.ms)

Offsets of surviving records do not change; compaction only removes records.

How it works

• The log is split into a **head** (recent, uncompacted, includes the active segment) and a **tail** (cleaned).
• A background **log cleaner** thread rewrites older segments, keeping only the newest record per key.
• Settings:

cleanup.policy=compact
min.cleanable.dirty.ratio=0.5      # clean when 50% of the log is "dirty"
min.compaction.lag.ms=0            # minimum age before a record can be compacted
max.compaction.lag.ms=...          # force compaction within this time
delete.retention.ms=86400000       # how long tombstones stay visible
segment.ms=604800000

• cleanup.policy=compact,delete combines compaction with time-based deletion.

Guarantees

• A consumer reading from the beginning sees at least the final value of every key.
• Ordering of records is preserved.
• Consumers that keep up see every update before it is compacted (if they are faster than the compaction lag).

Built-in users

__consumer_offsets, __transaction_state, Kafka Streams changelogs and Kafka Connect config/offset/status topics are compacted.

Summary

Log compaction keeps the latest value per key (and removes deleted keys after tombstones), turning a topic into a durable, replayable current-state table.

Solution and explanation

Use compacted topics when the topic represents **current state per key**

1. **Entity state / reference data:** customer profiles, product catalogue, job postings, branch IFSC codes, exchange rates. New services can bootstrap the full current state by reading the topic from the start.
2. **Kafka Streams state stores:** changelogs for tables and aggregations (automatic).
3. **Caches and materialised views:** a service rebuilds its in-memory cache or search index from the topic after a restart.
4. **Configuration distribution:** feature flags or settings per key.
5. **CDC "latest row" topics:** database tables mirrored as one record per primary key.
6. **Offsets and checkpoints:** like __consumer_offsets.

Do not use compaction when

• Every event matters (payments, clicks, ledger entries, audit logs): compaction would delete history. Use time retention or infinite retention instead.
• Records have no key (compaction requires keys; produce fails for null keys on compacted topics).
• You need data to disappear after a fixed time (compaction alone keeps the latest value forever; use compact,delete).

Design tips

• Choose the key to be the entity id; the value is the full latest state (not a partial delta), so the latest record alone is enough.
• Use tombstones to delete entities (including for privacy requests).
• Tune min.compaction.lag.ms if consumers must see every intermediate update.

Example

Topic jobs.current (key = jobId, compacted): the search indexer and recommendation service each rebuild their state by reading from offset 0, getting exactly one latest record per live job.

Summary

Use compacted topics for latest-state data keyed by entity (profiles, catalogues, changelogs, caches), not for event histories where every record matters.

Solution and explanation

What it is

A tombstone is a record with a **key and a null value**. On a compacted topic it means "**delete this key**": the log cleaner removes older records for that key, and later removes the tombstone itself after delete.retention.ms.

producer.send(new ProducerRecord<>("customer-profiles", "C42", null));   // delete customer C42

Lifecycle

1. The tombstone is appended like any record.
2. During compaction, all earlier records for C42 are removed.
3. The tombstone stays for delete.retention.ms (default 24 h) so consumers that read the topic can see the deletion and remove C42 from their own stores.
4. After that, the tombstone is removed too, so the key leaves no trace.

A consumer offline for longer than delete.retention.ms may miss the deletion; size this setting to your consumers' maximum downtime.

Where tombstones appear

• **Kafka Streams:** a KTable treats null values as deletes and emits tombstones to changelogs.
• **Debezium CDC:** a database DELETE produces a delete event followed by a tombstone so the compacted topic drops the row.
• **Sink connectors:** e.g. Elasticsearch with behavior.on.null.values=delete deletes the document.
• **Privacy:** removing a user's data from compacted topics on request (GDPR, India's DPDP Act).

Consumer handling

Code must handle null values: deserialisers and listeners should not crash on them.

@KafkaListener(topics = "customer-profiles")
public void on(ConsumerRecord<String, Customer> r) {
    if (r.value() == null) cache.remove(r.key()); else cache.put(r.key(), r.value());
}

Summary

A tombstone is a key with a null value that deletes the key from a compacted topic, kept for delete.retention.ms so consumers can apply the deletion.