Topology vs Processor #
When we build stream processing applications using Kafka Streams, all the business logic we write is ultimately translated into a structured computation flow graph. This structure is known as the Processor Topology. Understanding how this topology is designed, internally allocated, and how we interact with it through two main interfaces — the high-level DSL API (Domain Specific Language) and the low-level Processor API (PAPI) — is the key to building reliable and optimal stream processing systems. Through this article, we’ll thoroughly unpack the internal structure of Processor Topologies, the role differences of the three computation node types, an in-depth DSL vs Processor API comparison, and how to correctly write custom stateful processors in production.
Processor Topology Structure: Directed Acyclic Graph (DAG) #
Architecturally, a Processor Topology is a DAG (Directed Acyclic Graph) describing the data flow from Kafka input topics, through a series of logic processing nodes, ending at Kafka output topics.
Inside this graph, data flows as key-value records. Each point or node in the graph is called a Processor Node, and each directed line represents a data flow path.
The Three Main Computation Node Types #
Inside a Kafka Streams topology, there are three special node types with strictly defined responsibilities:
1. Source Processor #
A Source Processor acts as the data flow entrance into our topology.
- Duty: Subscribes to one or several physical Kafka topics, asynchronously receives data from Brokers, deserializes keys and values from binary format (
byte[]) into structured Java objects using serializer/deserializer (Serde), and forwards that data to downstream child nodes. - Source nodes have no parent nodes inside the topology.
2. Stream Processor (Processing Node) #
Stream Processors represent where actual data transformations happen.
- Duty: Receives records from parent nodes, applies transformation functions (stateless like
mapandfilter, or stateful likejoinandaggregate), and optionally forwards transformation results to the next downstream child nodes. - Processing nodes can access one or several local State Stores (RocksDB) to maintain temporary state.
3. Sink Processor (Outlet Node) #
Sink Processors act as the data flow exit from our topology.
- Duty: Receives records from parent nodes, re-serializes Java objects into binary format (
byte[]), and writes those final result data to physical destination Kafka topics on brokers. - Sink nodes have no child nodes inside the topology.
flowchart TD
subgraph TOPOLOGY["Processor Topology (DAG)"]
direction TB
Source["Source Node (Topic: raw-clicks)"]
ProcFilter["Stateless Processor (Filter Bots)"]
ProcCount["Stateful Processor (Count Clicks)"]
Store[("State Store (RocksDB)")]
Sink["Sink Node (Topic: aggregated-clicks)"]
Source --> ProcFilter
ProcFilter --> ProcCount
ProcCount <--> Store
ProcCount --> Sink
end
style Source stroke:#0288d1,stroke-width:2px
style ProcFilter stroke:#388e3c,stroke-width:2px
style ProcCount stroke:#388e3c,stroke-width:2px
style Store stroke:#f57c00,stroke-width:2px
style Sink stroke:#d32f2f,stroke-width:2pxDSL API vs Processor API (PAPI) #
Kafka Streams provides two different approaches for developers to define Processor Topologies. Choosing between them is a crucial design decision.
1. DSL API (Domain Specific Language) #
The DSL API is a high-level functional declarative programming interface. It’s the most common and recommended way for most ordinary use cases.
- Declarative: We specify what we want to do to the data stream (e.g.,
stream.filter(...).map(...).to(...)), and Kafka Streams automatically translates and optimizes those statements into a physical topology DAG in the background. - Rich Built-in Features: Provides out-of-the-box operators like mapping (
map,mapValues), filtering (filter), joining (join), grouping (groupBy), windowing, and direct topic writing (to). - Disadvantages: Limited low-level control customization. We don’t have direct access to record metadata (like offsets, physical timestamps, or Kafka headers) in detail, and we can’t manually schedule state store commits per event.
2. Processor API (PAPI) #
The Processor API is a low-level imperative programming interface. PAPI gives developers total control over the computation flow.
- Imperative: We must manually define the topology by explicitly adding sources, processors, sinks, and state stores to the
Topologyobject, then write processing logic procedurally line by line inside theprocess()method. - Maximum Control: We have full access to the
ProcessorContextto read offsets, timestamps, headers, dynamically emit data downstream (context.forward()), and schedule periodic executions usingPunctuator(e.g., triggering calculations every 5 minutes of system time or event time). - Disadvantages: We must write far more boilerplate code to handle serialization, manual state store management, and error handling.
Implementation Code: Anti-Pattern vs Stateful Solutions #
To understand the concrete differences between both APIs and how to safely handle topologies, let’s review a real use case example.
Use Case #
We want to monitor credit card transaction streams (credit-card-events). If a user performs consecutive transactions with identical amounts within less than 3 seconds (an indication of cashier system failures or card double taps), we want to filter out the second transaction event and send it to a special alert topic (suspicious-transactions).
Anti-Pattern: Mixing Ad-Hoc Stateful Logic in the DSL #
Trying to build complex stateful logic (storing last transactions per user) using stateless DSL operators like filter with external static HashMap variables is a dangerous anti-pattern.
// ANTI-PATTERN: Trying to manage last transaction state using static variables in a DSL filter.
// ✗ Not safe from crash failures, triggers multi-thread race conditions, and damages horizontal scaling.
public class DangerousDslStateStore {
// ✗ DANGER: A static HashMap is unprotected from crashes and not synchronized across threads/instances
private static final Map<String, Transaction> lastTransactions = new ConcurrentHashMap<>();
public static void buildTopology(StreamsBuilder builder) {
builder.<String, String>stream("credit-card-events")
.filter((userId, transactionJson) -> {
Transaction currentTx = parseJson(transactionJson);
Transaction lastTx = lastTransactions.get(userId);
if (lastTx != null) {
long diffSeconds = (currentTx.timestamp - lastTx.timestamp) / 1000;
// Double transaction detection filter within 3 seconds
if (currentTx.amount == lastTx.amount && diffSeconds < 3) {
// ✗ State modification without Changelog backup
lastTransactions.put(userId, currentTx);
return true; // Pass through as a suspicious transaction
}
}
lastTransactions.put(userId, currentTx);
return false;
})
.to("suspicious-transactions");
}
private static Transaction parseJson(String json) { return new Transaction(); }
static class Transaction { double amount; long timestamp; }
}
Practical Solution 1: Using the Processor API (PAPI) Cleanly #
To solve this challenge safely from server failures and ready to scale, we must use the Processor API with officially managed RocksDB state stores.
// CORRECT: Using the Processor API to safely manage state with local RocksDB and Changelog backups.
// ✓ Safe from data loss when containers die, free from thread concurrency issues.
public class TransactionDoubleTapProcessor implements Processor<String, String, String, String> {
private ProcessorContext<String, String> context;
private KeyValueStore<String, String> stateStore;
@Override
public void init(ProcessorContext<String, String> context) {
this.context = context;
// ✓ Getting a reference to the local RocksDB State Store registered in the topology
this.stateStore = context.getStateStore("last-transaction-store");
}
@Override
public void process(Record<String, String> record) {
String userId = record.key();
String currentTxJson = record.value();
// Quickly reading the user's last transaction from local RocksDB
String lastTxJson = stateStore.get(userId);
if (lastTxJson != null) {
double currentAmount = parseAmount(currentTxJson);
double lastAmount = parseAmount(lastTxJson);
long currentTime = record.timestamp();
long lastTime = parseTimestamp(lastTxJson);
long diffMs = currentTime - lastTime;
// ✓ Detecting double taps (identical value transactions within less than 3 seconds)
if (currentAmount == lastAmount && diffMs < 3000) {
// Emit this suspicious transaction to the downstream node
context.forward(record);
// Update the last transaction state in the RocksDB database
stateStore.put(userId, currentTxJson);
return;
}
}
// Store the latest transaction in RocksDB
stateStore.put(userId, currentTxJson);
}
@Override
public void close() {
// Cleanup logic if there are open external resources
}
private double parseAmount(String json) { return 50.0; }
private long parseTimestamp(String json) { return System.currentTimeMillis(); }
}
Physical Topology Configuration for the Processor API #
After writing the processor class above, we must explicitly assemble its physical topology using the built-in Topology object:
// CORRECT: Explicitly assembling the physical topology to register the Custom Processor and State Store.
public class TransactionTopologyBuilder {
public static Topology build() {
Topology topology = new Topology();
// 1. Add a Source Node to read the physical topic from the Broker
topology.addSource("TransactionSource", "credit-card-events");
// 2. Add the Custom Stateful Processor into the topology
topology.addProcessor(
"DoubleTapProcessor",
TransactionDoubleTapProcessor::new,
"TransactionSource" // Specifying its parent (Source Node)
);
// 3. Configure the local RocksDB State Store
StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("last-transaction-store"), // Persistent RocksDB
Serdes.String(),
Serdes.String()
);
// 4. Register the State Store to the topology and bind it with the processor needing it
topology.addStateStore(storeBuilder, "DoubleTapProcessor");
// 5. Add a Sink Node to write passed data to the suspicious-transactions output topic
topology.addSink(
"AlertSink",
"suspicious-transactions",
"DoubleTapProcessor" // Specifying its parent (Custom Processor Node)
);
return topology;
}
}
Practical Solution 2: The Hybrid Approach (DSL with process())
#
If we want to keep the DSL API convenience but need the stateful control power of the Processor API for certain logic parts, we can use the hybrid approach through the process() operator on KStreams.
// CORRECT: The Hybrid Approach using KStream.process() inside the DSL API.
// ✓ Combining DSL declarative ease with low-level custom processor flexibility.
public class HybridTopologyExample {
public static void buildHybrid(StreamsBuilder builder) {
// Configure the local RocksDB State Store
StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("last-transaction-store"),
Serdes.String(),
Serdes.String()
);
// Register the storeBuilder globally to the StreamsBuilder
builder.addStateStore(storeBuilder);
KStream<String, String> stream = builder.stream("credit-card-events");
// ✓ Using the process() operator to jump from DSL to PAPI transparently
stream
.process(
TransactionDoubleTapProcessor::new,
"last-transaction-store" // Associating the RocksDB store to the processor instance
)
.to("suspicious-transactions");
}
}
When to Choose DSL and When to Switch to PAPI? #
To ease decision-making for developer teams, here’s the interface selection decision flow matrix:
STILL USE THE DSL API IF:
✓ Our data processing needs are generic (like doing simple map, filter, or joins).
✓ We want concise, quickly readable, and easy-to-maintain code.
✓ Our stateful operations can be directly mapped with built-in aggregate(), count(), or reduce() functions.
SWITCH TO THE PROCESSOR API (PAPI) IF:
✗ We need to dynamically emit data to several different output topics based on runtime logic conditions (Dynamic Routing).
✗ We need direct access to low-level metadata like Record Headers, Partition metadata, or Offsets.
✗ We need to trigger periodic actions (like clearing expired old state) based on system/event time intervals using Punctuation.
✗ We want to custom-modify local RocksDB state store behavior beyond the built-in APIs.
Summary #
- Processor Topology — A directed acyclic graph (DAG) defining asynchronous data flows from input topics, through processing nodes, to output topics.
- Three Node Types — Topologies consist of Source Nodes (deserialization & input), Processor Nodes (transformation logic), and Sink Nodes (serialization & output).
- DSL API — A high-level declarative interface rich with built-in functions, perfect for standard stream processing requiring fast implementation.
- Processor API (PAPI) — A low-level imperative interface giving full control over state management, record metadata, and execution scheduling.
ProcessorContext— The connector object in PAPI providing access to runtime metadata, state stores, and downstream data forwarding mechanisms (forward()).Punctuator— The scheduling component in PAPI for periodically executing custom functions based on system time (Wall-clock Time) or data arrival time (Stream Time).