Kafka Streams DSL vs Processor API: When to Use Each

Introduction

Apache Kafka Streams exposes two programming models for building stream processing applications: the high-level DSL and the low-level Processor API. The DSL is declarative and functional, building the topology for you. The Processor API exposes the topology graph directly, giving you control over record forwarding, state-store registration, and punctuation (scheduled callbacks).

The two are not separate runtimes. The DSL compiles down to a graph of processor nodes—the same Topology object the Processor API builds by hand—so both share identical partition assignment, state restoration, and fault tolerance. That shared foundation is what makes the hybrid model practical: you can write the bulk of a topology in the DSL and drop into the Processor API only where the DSL cannot express your logic, via KStream#process and KStream#processValues.

This article covers the operational and developer-ergonomics trade-offs, gives runnable code for both APIs against the modern (Kafka 3.0+) interfaces, and sets out a decision framework. For teams running Stream Processing pipelines—including Stateful Stream Processing with Kafka Streams—the choice mostly affects code maintainability and the granularity of control over state and record handling, not the runtime behaviour.

API version note. All examples target the type-safe Processor API introduced in KIP-478 and shipped in Kafka Streams 3.0. The processor implements Processor<KIn, VIn, KOut, VOut> directly (the old AbstractProcessor was removed in 3.0). process() takes a Record, the ProcessorContext is stored from init() and called as a field (context.forward(...), not context().forward(...)), and forward takes an immutable Record.

Architectural Foundations: How Both APIs Construct Topologies

Both APIs produce a Topology: a directed acyclic graph of source, processor, and sink nodes connected by stream edges. The runtime executes that graph the same way regardless of how it was defined.

The Processor API: Explicit Topology Construction

The Processor API requires you to define source, processor, and sink nodes explicitly and wire them together by name. The supplier passed to addProcessor is a ProcessorSupplier, so a lambda returning a fresh Processor instance works:

Topology topology = new Topology();

topology.addSource("Source", "input-topic")
        .addProcessor("Transform", () -> new Processor<String, String, String, String>() {
            private ProcessorContext<String, String> context;

            @Override
            public void init(ProcessorContext<String, String> context) {
                this.context = context;
            }

            @Override
            public void process(Record<String, String> record) {
                String transformed = record.value().toUpperCase();
                context.forward(record.withValue(transformed));
            }

            @Override
            public void close() {
            }
        }, "Source")
        .addSink("Sink", "output-topic", "Transform");

Two details matter for correctness against the 3.0+ API. First, the context is captured in init() and forwarding goes through the stored field context.forward(...)—there is no context() accessor method on this interface. Second, Record is immutable; record.withValue(...) returns a new copy rather than mutating in place.

The payoff for this verbosity is naming control. Because you name every node, the metrics emitted per processor (process-rate, process-latency-avg, and so on, tagged by processor-node-id) map to names you chose, which makes it straightforward to pinpoint which node is the bottleneck on a dashboard.

The DSL: Declarative Topology Construction

The DSL expresses the same logic through a fluent API and builds the topology for you:

StreamsBuilder builder = new StreamsBuilder();
builder.stream("input-topic")
       .mapValues(value -> value.toUpperCase())
       .to("output-topic");

The trade-off is opacity. Run Topology#describe() and you will see generated node names like KSTREAM-MAPVALUES-0000000001, which carry no domain meaning. You can override most of them with the Named and Materialized parameters; the Kafka Streams developer guide documents naming for operators and state stores.

Combining Both APIs in a Single Application

You can mix both APIs in one StreamsBuilder. KStream#process (and the value-only KStream#processValues) injects a Processor API node into a DSL topology. As of KIP-820 (Kafka 3.3), process returns a KStream, so you can continue the DSL chain afterward; the older transform/transformValues/flatTransform family and the void-returning process overloads are deprecated in their favour.

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input-topic");

// DSL for the standard transformation
KStream<String, String> enriched = stream.mapValues(v -> v.toUpperCase());

// Processor API node for logic that needs record metadata
KStream<String, String> withTimestamp = enriched.process(() ->
    new Processor<String, String, String, String>() {
        private ProcessorContext<String, String> context;

        @Override
        public void init(ProcessorContext<String, String> context) {
            this.context = context;
        }

        @Override
        public void process(Record<String, String> record) {
            long timestamp = record.timestamp();
            String value = record.value() + "|" + timestamp;
            context.forward(record.withValue(value));
        }

        @Override
        public void close() {
        }
    });

withTimestamp.to("output-topic");

Record timestamp comes from record.timestamp() on the Record passed to process(). Topic, partition, and offset (when available) come from context.recordMetadata(), which returns an Optional<RecordMetadata>—it is empty for records emitted from a punctuator, since those have no source record.

Operational Characteristics: Performance, State, and Resources

State Store Management

The DSL creates and manages state stores for you. A groupByKey().aggregate() provisions a RocksDB-backed store (in-memory if you ask for it via Materialized), creates the changelog topic, and restores state on rebalance—no wiring required. The cost is reduced control over store configuration.

With the Processor API you register stores explicitly and access them by name:

StoreBuilder<KeyValueStore<String, Long>> storeBuilder =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("my-store"),
        Serdes.String(),
        Serdes.Long()
    );

Topology topology = new Topology();
topology.addSource("Source", "input-topic")
        .addProcessor("Counter", () -> new Processor<String, String, String, Long>() {
            private KeyValueStore<String, Long> store;

            @Override
            public void init(ProcessorContext<String, Long> context) {
                store = context.getStateStore("my-store");
            }

            @Override
            public void process(Record<String, String> record) {
                Long current = store.get(record.key());
                store.put(record.key(), current == null ? 1L : current + 1);
            }

            @Override
            public void close() {
            }
        }, "Source");

topology.addStateStore(storeBuilder, "Counter");

Note the explicit addStateStore(storeBuilder, "Counter"): a Processor API store must be added to the topology and connected to the processor(s) that use it, otherwise getStateStore fails at startup. This explicit wiring is also what lets two processors share one store—something the DSL never does, since each stateful DSL operator owns an isolated store.

RocksDB tuning (block cache, write buffers, compaction style) is available to both APIs through RocksDBConfigSetter, configured with StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG. The Processor API’s advantage here is per-store control via custom KeyValueBytesStoreSupplier implementations, not exclusive access to RocksDB settings.

Record Metadata Access and Custom Scheduling

The Processor API supports punctuation: a scheduled callback driven by either wall-clock time or stream time. schedule takes a Duration, a PunctuationType, and a Punctuator, and returns a Cancellable:

public class ScheduledCleanupProcessor implements Processor<String, String, Void, Void> {
    private KeyValueStore<String, String> store;
    private ProcessorContext<Void, Void> context;

    @Override
    public void init(ProcessorContext<Void, Void> context) {
        this.context = context;
        this.store = context.getStateStore("my-store");
        context.schedule(Duration.ofMinutes(5), PunctuationType.WALL_CLOCK_TIME, timestamp -> {
            try (KeyValueIterator<String, String> iter = store.all()) {
                while (iter.hasNext()) {
                    KeyValue<String, String> entry = iter.next();
                    if (isExpired(entry.value)) {
                        store.delete(entry.key);
                    }
                }
            }
        });
    }

    @Override
    public void process(Record<String, String> record) {
        store.put(record.key(), record.value());
    }

    @Override
    public void close() {
    }

    private boolean isExpired(String value) {
        // Custom expiration logic
        return false;
    }
}

WALL_CLOCK_TIME fires roughly on a real-time interval (only when the consumer poll loop runs, so it is approximate); STREAM_TIME advances with the timestamps of incoming records and does not fire while a partition is idle. There is no DSL equivalent, which makes punctuation a common reason to reach for the Processor API: TTL-based eviction, periodic flushes to an external sink, or heartbeat emission.

Performance Overhead

The DSL’s abstraction over the Processor API is thin; per-record overhead is negligible. The meaningful differences are structural. The DSL applies its own optimizations—for example, enabling StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG = OPTIMIZE lets it reuse a source topic as a changelog and collapse repartition topics, reducing broker-side traffic. The Processor API, conversely, lets you fold several logical steps (deserialize, route, enrich) into one node and avoid the intermediate repartition/serialization that an equivalent chain of DSL operators might introduce.

When to Use the DSL: Standard Stream Processing Patterns

The DSL excels at the common patterns—filter, map, group, aggregate, join, window—which cover the large majority of production pipelines with far less code.

Rapid Development and Reduced Boilerplate

A typical enrichment pipeline—read, filter, join with a reference table, aggregate, write—is concise and self-documenting in the DSL:

StreamsBuilder builder = new StreamsBuilder();

KStream<String, Order> orders = builder.stream("orders-topic",
    Consumed.with(Serdes.String(), orderSerde));

KTable<String, Customer> customers = builder.table("customers-topic",
    Consumed.with(Serdes.String(), customerSerde));

orders.filter((key, order) -> order.getAmount() > 0)
      .join(customers, Order::getCustomerId,
          (order, customer) -> new EnrichedOrder(order, customer))
      .groupByKey(Grouped.with(Serdes.String(), enrichedOrderSerde))
      .aggregate(OrderSummary::new,
          (key, enriched, summary) -> summary.add(enriched),
          Materialized.with(Serdes.String(), summarySerde))
      .toStream()
      .to("enriched-orders-topic");

The join(KTable, KeyValueMapper, ValueJoiner) overload above is a foreign-key stream-table join (the mapper extracts the join key from the order). The equivalent Processor API implementation would be far longer: manual store registration, a join buffer, explicit forwarding, and careful node wiring.

Built-in Support for Common Operations

Windowing is the clearest example. The DSL offers tumbling, hopping, session, and sliding windows directly:

orders.groupByKey()
      .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
      .aggregate(OrderSummary::new,
          (key, order, summary) -> summary.add(order),
          Materialized.with(Serdes.String(), summarySerde));

TimeWindows.ofSizeWithNoGrace(Duration) is the modern factory (Kafka 3.3+); it replaced the deprecated TimeWindows.of(Duration), whose default grace period was easy to misjudge. Use ofSizeAndGrace(size, grace) when you need to admit late records. Re-implementing session windows by hand in the Processor API means tracking session boundaries, merging overlapping sessions, and managing expiry—work the DSL does correctly out of the box.

Testing and Maintainability

DSL operators are largely pure functions, which makes the transformation logic easy to extract and unit-test. The Unit Testing Kafka Streams Topologies with TopologyTestDriver approach exercises a full topology in-process, with no broker.

When to Use the Processor API: Advanced Requirements

Reach for the Processor API when requirements exceed what the DSL can express—custom state, complex routing, external integration, or scheduled work.

Custom State Store Strategies

When your state does not fit a single key-value store per operator—shared stores, secondary indexes, a custom StateStore backend—the Processor API is the only option. Multiple processors can attach to one registered store, supporting patterns where one node writes and another reads it for a different purpose. (Note that any store you want to survive restarts still needs a changelog or persistentKeyValueStore; an “external Redis” store, by contrast, lives outside Kafka’s fault-tolerance model and you own its consistency.)

Complex Record Routing and Multi-Output Topologies

When you need to route a record to one of several named downstream processors based on content, the Processor API forwards to a child by name. In the 3.0+ API the second forward argument is the child node’s String name—there is no To overload on org.apache.kafka.streams.processor.api.ProcessorContext:

public class ContentRouter implements Processor<String, byte[], String, byte[]> {
    private ProcessorContext<String, byte[]> context;

    @Override
    public void init(ProcessorContext<String, byte[]> context) {
        this.context = context;
    }

    @Override
    public void process(Record<String, byte[]> record) {
        Header header = record.headers().lastHeader("Content-Type");
        String contentType = header == null ? "" : new String(header.value());

        switch (contentType) {
            case "application/json":
                context.forward(record, "json-processor");
                break;
            case "application/avro":
                context.forward(record, "avro-processor");
                break;
            default:
                context.forward(record, "dlq-processor");
        }
    }

    @Override
    public void close() {
    }
}

The named children (json-processor, avro-processor, dlq-processor) must exist as child nodes wired beneath this processor in the topology. The DSL’s branching equivalent is KStream#split(), which returns a BranchedKStream (the older KStream#branch has been deprecated since Kafka 2.8). split is fine for value-based predicates, but the Processor API version lets the targets be stateful processors with their own stores and chosen node names.

Integration with External Systems

The Processor API’s init/process/close lifecycle gives you hooks to manage connection pools and shut down cleanly:

public class ExternalEnrichmentProcessor implements Processor<String, String, String, String> {
    private HttpClient httpClient;
    private ProcessorContext<String, String> context;

    @Override
    public void init(ProcessorContext<String, String> context) {
        this.context = context;
        this.httpClient = HttpClient.newBuilder()
            .connectTimeout(Duration.ofSeconds(5))
            .build();
    }

    @Override
    public void process(Record<String, String> record) {
        // Blocking call: Streams processing is single-threaded per task,
        // so synchronous I/O here serializes the partition. Bound it with
        // a request timeout, and prefer batching/caching for hot keys.
        try {
            HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create("https://api.example.com/enrich/" + record.key()))
                .timeout(Duration.ofSeconds(2))
                .GET()
                .build();

            HttpResponse<String> response =
                httpClient.send(request, HttpResponse.BodyHandlers.ofString());
            context.forward(record.withValue(record.value() + "|" + response.body()));
        } catch (IOException | InterruptedException e) {
            if (e instanceof InterruptedException) Thread.currentThread().interrupt();
            // Route to a DLQ child or rethrow per your error policy.
        }
    }

    @Override
    public void close() {
        httpClient.close(); // HttpClient is AutoCloseable as of Java 21
    }
}

Two correctness points. First, prefer a synchronous call here: a Streams task processes records single-threaded, and the immutability/commit semantics mean you should forward from the same process() invocation rather than from an async callback that may run after the offset has been committed. If you do need async I/O, buffer the in-flight work in a state store and emit completions from a punctuator. Second, HttpClient.close() is valid only on Java 21+, where HttpClient became AutoCloseable; on earlier JDKs the instance has no close() and is reclaimed by GC (its executor threads are daemons). Size the request timeout deliberately—an unbounded external call will stall the entire partition.

Custom Punctuation and Time-Based Logic

Punctuation, shown earlier, has no DSL equivalent. Typical uses:

• Emitting heartbeat records downstream
• Flushing aggregated data to an external sink on a schedule
• Custom windowing strategies that do not fit the built-in window types
• Periodic state-store cleanup or eviction

Decision Framework

Start with the DSL by Default

For new applications, start with the DSL: it is concise, applies its own optimizations, and manages state for you. It covers the bulk of production use cases—stateless transforms, aggregations and reductions, stream-table and stream-stream joins, all four window types, and branching/merging. If the whole topology fits in DSL operators, keep it in the DSL; the result is easier to read, test, and hand off.

Drop into the Processor API When Necessary

Introduce Processor API nodes for the cases the DSL cannot express:

1. Custom or shared state stores — multiple processors on one store, or a non-RocksDB backend.
2. Record metadata — logic that depends on timestamp, headers, or context.recordMetadata() (topic/partition/offset).
3. Complex routing — forwarding to multiple named children on dynamic conditions.
4. Punctuation — periodic wall-clock or stream-time actions.
5. External integration — managed connection lifecycles, retries, bounded blocking I/O.
6. Hot paths — when profiling shows a hand-written node removes a repartition or redundant serialization step.

The integration points for the hybrid model are KStream#process(ProcessorSupplier, ...) and KStream#processValues(FixedKeyProcessorSupplier, ...). processValues is value-only and forbids key changes (it hands you a FixedKeyRecord), which preserves partitioning and avoids an unnecessary repartition; use it whenever the key does not change.

Operational Considerations for Production

• Observability — explicit node names produce interpretable per-node metrics. With the DSL, name operators and stores via Named and Materialized so dashboards do not show KSTREAM-*-0000000007.
• State-store sizing — defaults may not suit large state. Tune RocksDB via a RocksDBConfigSetter (available to both APIs) and consider custom store suppliers in the Processor API.
• Rebalancing — both APIs use the same consumer-group protocol and restore state identically. Enable StreamsConfig.OPTIMIZE and standby replicas to reduce restore time.
• Upgrade compatibility — the DSL surface is stable, but topology-changing optimizations and operator additions can shift generated node and changelog names; pin names with Named/Materialized so a library upgrade does not orphan your changelog topics. The Kafka Streams upgrade guide documents version-specific behaviour.

Testing Strategies for Both APIs

TopologyTestDriver runs a full topology in-process with no broker and works for DSL and Processor API topologies alike. It requires a Properties with at least APPLICATION_ID_CONFIG and BOOTSTRAP_SERVERS_CONFIG (the bootstrap value is never dialled, so any placeholder is fine).

Testing DSL Topologies

StreamsBuilder builder = new StreamsBuilder();
builder.stream("input", Consumed.with(Serdes.String(), Serdes.String()))
       .filter((k, v) -> v.length() > 0)
       .mapValues(String::toUpperCase)
       .to("output");

Topology topology = builder.build();
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");

try (TopologyTestDriver testDriver = new TopologyTestDriver(topology, props)) {
    TestInputTopic<String, String> inputTopic = testDriver.createInputTopic(
        "input", Serdes.String().serializer(), Serdes.String().serializer());
    TestOutputTopic<String, String> outputTopic = testDriver.createOutputTopic(
        "output", Serdes.String().deserializer(), Serdes.String().deserializer());

    inputTopic.pipeInput("key1", "hello");
    assertEquals("HELLO", outputTopic.readValue());
}

These tests are fast and deterministic and run in CI without infrastructure. The driver also exposes state stores for assertions after a run.

Testing Processor API Topologies

The same driver works; you typically also assert on store contents. Retrieve the store with getKeyValueStore after constructing the driver (the store does not exist until the topology is initialized):

Topology topology = new Topology();
topology.addSource("Source", "input")
        .addProcessor("Process", MyCustomProcessor::new, "Source")
        .addSink("Sink", "output", "Process");
topology.addStateStore(storeBuilder, "Process"); // wire the store to the processor

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");

try (TopologyTestDriver testDriver = new TopologyTestDriver(topology, props)) {
    TestInputTopic<String, String> input = testDriver.createInputTopic(
        "input", Serdes.String().serializer(), Serdes.String().serializer());
    input.pipeInput("expected-key", "value");

    KeyValueStore<String, Long> store = testDriver.getKeyValueStore("my-store");
    assertNotNull(store.get("expected-key"));
}

The Unit Testing Kafka Streams Topologies with TopologyTestDriver guide covers the full testing surface—inputs, outputs, state, and time advancement.

Conclusion

The DSL and Processor API are complementary, not competing. The DSL is the productive default for standard patterns; the Processor API is the escape hatch for custom state, routing, scheduling, and external integration. They share a runtime, a fault-tolerance model, and the same scaling behaviour—so the real decision is about ergonomics, maintainability, and how much control you need over state and record handling.

Default to the DSL, and inject Processor API nodes through process/processValues exactly where the DSL falls short. That hybrid—DSL for the standard majority of the topology, Processor API for the parts that need it—gives you a maintainable codebase that can still express the full complexity of your domain. For broader context, see Stream Processing.

Related Pages

• Windowed Aggregations and Joins in ksqlDB