Apache Kafka: Streams, Connect & Best Practices

Kafka Streams

Kafka Streams is a client library for building real-time stream processing applications. It runs inside your application — no separate cluster needed.

// Build stream processing topology
StreamsBuilder builder = new StreamsBuilder();

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

// Filter, transform
KStream<String, Order> parsedOrders = orders
    .filter((key, value) -> value != null)
    .mapValues(value -> parseOrder(value));

// Aggregate per user in a 5-minute window
parsedOrders
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .aggregate(
        () -> 0.0,
        (key, order, total) -> total + order.getAmount(),
        Materialized.as("order-totals-store")
    )
    .toStream()
    .to("order-totals-per-user");

// Join two streams (within 60s window)
KStream<String, Payment> payments = builder.stream("payments");
parsedOrders.join(
    payments,
    (order, payment) -> new FulfilledOrder(order, payment),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofSeconds(60))
).to("fulfilled-orders");

// Interactive queries (query materialized state)
KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();
ReadOnlyKeyValueStore<String, Double> store =
    streams.store(StoreQueryParameters.fromNameAndType("order-totals-store",
        QueryableStoreTypes.keyValueStore()));
Double total = store.get("user-123");

Kafka Connect

Kafka Connect is a framework for streaming data between Kafka and external systems using reusable connectors.

// Source connector: PostgreSQL → Kafka (Debezium CDC)
{
  "name": "postgres-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "replicator",
    "database.password": "secret",
    "database.dbname": "mydb",
    "table.include.list": "public.orders,public.products",
    "topic.prefix": "cdc",
    "plugin.name": "pgoutput"
  }
}

// Sink connector: Kafka → Elasticsearch
{
  "name": "elasticsearch-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "1",
    "topics": "orders",
    "connection.url": "http://elasticsearch:9200",
    "type.name": "_doc",
    "key.ignore": "false",
    "schema.ignore": "true"
  }
}

# Manage connectors via REST API
curl -X POST http://connect:8083/connectors   -H "Content-Type: application/json"   -d @connector-config.json

curl http://connect:8083/connectors                      # list
curl http://connect:8083/connectors/postgres-source/status  # status
curl -X DELETE http://connect:8083/connectors/postgres-source

Best Practices

• Partition count: start with num_consumers * 2; can only increase partitions (never decrease).

• Replication factor 3: tolerate 1 broker failure without data loss (acks=all + min.insync.replicas=2).

• Use message keys for ordering guarantees — all records with the same key go to the same partition.

• Schema Registry (Confluent): enforce Avro/JSON Schema, enable schema evolution with compatibility rules.

• Monitor consumer lag: kafka-consumer-groups.sh --describe, or use Burrow/Kafdrop/Cruise Control.

• Log compaction for state topics: retain only latest value per key (good for CDC and lookup tables).

• Dead Letter Queue (DLQ): route failed/unparseable messages to a separate topic for inspection.

• Idempotent consumers: always design consumers to handle duplicate messages safely.