Definition
- Kafka Streams -- a client library for building stream-processing applications directly on Kafka; it runs inside your application, not a separate cluster.
- Topology -- a directed graph of source nodes (read from topic), processor nodes (transform), and sink nodes (write to topic).
- mapValues -- a stateless transformation that changes only the value, keeping the key and partition unchanged, so no repartition is needed.
- Application ID -- doubles as the consumer group id and names internal state and changelog topics; each distinct app must have a unique id.
- Scaling -- scales by partition just like a consumer group; running more instances divides the input partitions among them.
Imagine a conveyor belt at National Bank of India that carries transaction descriptions. Kafka Streams is a small worker standing by the belt who reads each slip, rewrites the description in capital letters for a standardized ledger, and places it on an outgoing belt. The worker lives inside your own office, not a separate factory, and if you add more workers they automatically split the belts between them.
Understanding the Concepts
What is Kafka Streams?
Kafka Streams is a library, not a server, so you add it as a dependency and run it inside your Java application. You configure it with an application id, bootstrap servers, and default serdes (serializer-deserializer pairs). The application id names the consumer group and the internal topics.
package com.bank.streams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.serialization.Serdes;
import java.util.Properties;
public class StreamsConfigDemo {
public static Properties createConfig() {
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "transaction-normalizer");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
System.out.println("Kafka Streams config created");
System.out.println(" Application ID: transaction-normalizer");
System.out.println(" Default serdes: String/String");
return props;
}
public static void main(String[] args) {
createConfig();
}
}
Expected output:
Kafka Streams config created
Application ID: transaction-normalizer
Default serdes: String/String
How do you build the topology?
You describe the processing as a topology using the StreamsBuilder DSL. builder.stream("input-topic") creates a KStream. The mapValues transformation changes only the value, and .to("output-topic") writes the result.
package com.bank.streams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.Topology;
public class UppercaseTopology {
public static Topology build() {
StreamsBuilder builder = new StreamsBuilder();
// Source: read from raw-transactions topic
KStream<String, String> rawStream = builder.stream("raw-transactions");
// Transform: uppercase the transaction description
KStream<String, String> normalizedStream = rawStream
.mapValues(value -> value.toUpperCase());
// Sink: write to normalized-transactions topic
normalizedStream.to("normalized-transactions");
Topology topology = builder.build();
System.out.println("Topology built: raw-transactions -> uppercase -> normalized-transactions");
return topology;
}
public static void main(String[] args) {
build();
}
}
Expected output:
Topology built: raw-transactions -> uppercase -> normalized-transactions
How do you write the complete application?
The complete app creates the config, builds the topology, starts the KafkaStreams instance, and adds a shutdown hook.
package com.bank.streams;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.common.serialization.Serdes;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
public class TransactionNormalizerApp {
public static void main(String[] args) {
// Configuration
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "transaction-normalizer");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
// Topology
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> rawStream = builder.stream("raw-transactions");
rawStream.mapValues(String::toUpperCase).to("normalized-transactions");
// Start
KafkaStreams streams = new KafkaStreams(builder.build(), props);
CountDownLatch latch = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
streams.close();
latch.countDown();
System.out.println("Streams app shut down cleanly");
}));
try {
streams.start();
System.out.println("Transaction normalizer started");
System.out.println(" Input: raw-transactions");
System.out.println(" Output: normalized-transactions");
latch.await();
} catch (Exception e) {
System.exit(1);
}
}
}
Expected output:
Transaction normalizer started
Input: raw-transactions
Output: normalized-transactions
Why use mapValues instead of map?
mapValues transforms only the value and leaves the key and partition unchanged. If you used map and changed the key, Streams would repartition the data, which is more expensive and can reorder records.
| Operation | Changes Key? | Repartition? | Preserves Order? |
|---|---|---|---|
| mapValues | No | No | Yes |
| map | Yes | Yes (if key changes) | May reorder |
| flatMapValues | No | No | Yes |
| flatMap | Yes | Yes (if key changes) | May reorder |
package com.bank.streams;
public class MapVsMapValues {
public static void main(String[] args) {
// mapValues: only value changes, key stays, no repartition
System.out.println("mapValues: key=SAV2001, value='debit 50000' -> 'DEBIT 50000'");
System.out.println(" Same key, same partition, same order -- correct");
// map: key changes, triggers repartition
System.out.println("map: key=SAV2001 -> key=SAVINGS-SAV2001");
System.out.println(" New key, repartition needed, possible reorder -- expensive");
}
}
Expected output:
mapValues: key=SAV2001, value='debit 50000' -> 'DEBIT 50000'
Same key, same partition, same order -- correct
map: key=SAV2001 -> key=SAVINGS-SAV2001
New key, repartition needed, possible reorder -- expensive
How does Kafka Streams scale and recover?
A Streams app scales by partition. The input topic's partitions are divided among running instances of the same application id, exactly like a consumer group. If an instance dies, its partitions are reassigned to survivors.
6-partition input topic, 3 instances:
Instance 1: partitions 0, 1
Instance 2: partitions 2, 3
Instance 3: partitions 4, 5
If Instance 2 dies:
Instance 1: partitions 0, 1, 2
Instance 3: partitions 3, 4, 5
How do you test a Kafka Streams topology?
Use the TopologyTestDriver for unit testing without a running broker.
package com.bank.streams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;
public class TopologyTest {
public static void main(String[] args) {
// Build topology
StreamsBuilder builder = new StreamsBuilder();
builder.<String, String>stream("raw-transactions")
.mapValues(String::toUpperCase)
.to("normalized-transactions");
// Test config
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
try (TopologyTestDriver driver = new TopologyTestDriver(builder.build(), props)) {
TestInputTopic<String, String> input = driver.createInputTopic(
"raw-transactions", new StringSerializer(), new StringSerializer());
TestOutputTopic<String, String> output = driver.createOutputTopic(
"normalized-transactions", new StringDeserializer(), new StringDeserializer());
input.pipeInput("SAV2001", "debit 50000 for Rajesh");
String result = output.readValue();
System.out.println("Input: debit 50000 for Rajesh");
System.out.println("Output: " + result);
System.out.println("Test passed: " + "DEBIT 50000 FOR RAJESH".equals(result));
}
}
}
Expected output:
Input: debit 50000 for Rajesh
Output: DEBIT 50000 FOR RAJESH
Test passed: true
What about exactly-once processing in Kafka Streams?
Setting the processing guarantee to exactly_once_v2 makes the read-process-write atomic using Kafka transactions.
package com.bank.streams;
import org.apache.kafka.streams.StreamsConfig;
import java.util.Properties;
public class ExactlyOnceConfig {
public static void main(String[] args) {
Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");
System.out.println("Processing guarantee: exactly_once_v2");
System.out.println(" Read, transform, and write are atomic");
System.out.println(" Uses Kafka transactions under the hood");
}
}
Expected output:
Processing guarantee: exactly_once_v2
Read, transform, and write are atomic
Uses Kafka transactions under the hood
What should a senior engineer know?
For stateful operations like aggregations, Streams keeps local state stores backed by changelog topics. For our stateless uppercase app, failover is trivial because there is no state. Streams preserves per-key ordering because records for a key stay in one partition. The topology can be extracted and unit-tested with TopologyTestDriver without a running broker.
Visual Overview
Kafka Streams Topology
=======================
┌────────────────────────┐
│ Source Node │
│ raw-transactions │
│ (KStream) │
└───────────┬────────────┘
│
▼
┌────────────────────────┐
│ Processor Node │
│ mapValues(uppercase) │
│ (stateless transform) │
└───────────┬────────────┘
│
▼
┌────────────────────────┐
│ Sink Node │
│ normalized-transactions│
└────────────────────────┘
Scaling by Partition
=====================
Input: 6 partitions
┌───────────┐ ┌───────────┐ ┌───────────┐
│Instance 1 │ │Instance 2 │ │Instance 3 │
│ P0, P1 │ │ P2, P3 │ │ P4, P5 │
└───────────┘ └───────────┘ └───────────┘
Each instance processes its assigned partitions
Same application.id = same consumer group
Key Takeaway
Kafka Streams is a client library for stream processing that runs inside your own application rather than a separate cluster. To read a topic, uppercase values, and write to another topic, the whole data path is three steps: stream the input, mapValues with uppercase, and write to the output. Then you build the topology, create a KafkaStreams instance, and start it.
The important choice is mapValues instead of map, because mapValues transforms only the value and keeps the key unchanged, so there is no repartition and per-account ordering is preserved. The application scales by partition just like a consumer group.
In my National Bank of India project, transaction descriptions arrived in mixed case from different channels. A small Streams app normalized them with mapValues while keeping the account key, and I unit-tested the topology with TopologyTestDriver without a running broker.
Remember one sentence: Kafka Streams is a library, not a server, and mapValues transforms values without repartitioning, so stream, mapValues, and to is the complete stateless topology.
---