Kafka Streams & Connect

Kafka Streams: processing a topic without a separate cluster#

A consumer that reads a topic, transforms records, and writes to another topic is such a common pattern that Kafka ships a library for it: Kafka Streams. It's just a Java library your application depends on — there's no separate processing cluster to run, unlike Spark or Flink.

Here's a small topology that filters orders down to high-value ones and flags them, writing the results to a new topic:

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.KStream;

import java.util.Properties;

public class HighValueOrderFlagger {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

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

        KStream<String, String> highValueOrders = orders
            .filter((orderId, json) -> extractAmount(json) > 1000)
            .mapValues(json -> withPriorityFlag(json));

        highValueOrders.to("high-value-orders");

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-priority-flagging");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }

    private static double extractAmount(String json) { /* parse amount */ return 0; }
    private static String withPriorityFlag(String json) { /* add a field */ return json; }
}

builder.build() compiles the .filter().mapValues().to(...) chain into a topology — a graph of processing steps — which then runs continuously against the input topic once streams.start() is called. Two details matter for how this behaves operationally:

• APPLICATION_ID_CONFIG isn't just a label — it doubles as the underlying consumer group ID, and prefixes any internal topics Kafka Streams creates for itself (changelog topics backing any stateful operations like aggregations or joins, which this simple filter/map example doesn't use).
• Because it's built on the regular consumer API underneath, everything from the previous page still applies: partition count still caps parallelism, and a KafkaStreams instance restarted after a crash resumes from the last committed offset like any other consumer.

Kafka Connect: moving data in and out without writing a client#

Writing a one-off consumer to copy every row change from a database into Kafka, or to copy topic data out into a data warehouse, is exactly the kind of plumbing Kafka Connect exists to make declarative instead of hand-written:

• A source connector pulls data into Kafka from an external system — e.g., Debezium capturing row-level changes from a Postgres database's write-ahead log and publishing them as events.
• A sink connector pushes data out of Kafka into an external system — e.g., a JDBC sink connector writing consumed records into a reporting database table.

Connect runs as its own process (or cluster of processes) with a REST API for managing connectors — you post a JSON config, and it starts running:

{
  "name": "orders-jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://reporting-db:5432/analytics",
    "connection.user": "connect",
    "auto.create": "true",
    "insert.mode": "upsert",
    "pk.mode": "record_key"
  }
}

curl -X POST -H "Content-Type: application/json" \
  --data @orders-jdbc-sink.json \
  http://localhost:8083/connectors

tasks.max is the connector's parallelism knob — much like a Kafka Streams application, a sink connector consuming a topic is still bound by that topic's partition count, so tasks.max above the partition count buys you nothing.

When to reach for which#

• Custom business logic, joins, or aggregations over one or more topics → Kafka Streams (it's just a library in your own application).
• Moving data between Kafka and an external system with no transformation logic needed (a database, a search index, a data warehouse) → Kafka Connect (a config file, not custom code).
• Both can be used together: a source connector ingests raw change events, a Streams application enriches or aggregates them, and a sink connector exports the result.

Common mistakes#

• Writing a custom consumer application to do plain data movement (DB → Kafka, or Kafka → warehouse) that a well-maintained Connect connector already solves declaratively, with far less code to operate and debug.
• Setting tasks.max higher than the source topic's partition count and expecting more parallelism — the extra tasks simply have no partitions to be assigned.
• Forgetting that Kafka Streams' internal state (used for aggregations, joins, or windowing) is itself backed by a replicated Kafka changelog topic — the state isn't just in memory and disposable, it has the same replication/retention considerations as any other topic on the cluster.