The Basic Tech Info
Scaling Kafka Streams in Production: Partitioning, Tuning, and Monitoring

Scaling Kafka Streams in Production: Partitioning, Tuning, and Monitoring

Published:

Last Modified:

, ,

Moving a Kafka Streams app from local dev to production means dealing with skewed partitions, consumer lag, rebalances, and state store bloat. This guide is a field-tested playbook with runnable snippets: how to choose keys, increase partitions without downtime, tune Streams configs, enable static membership, and set up metrics you can trust.

The Production Playbook

Partitioning that actually scales

Pick a key that spreads load

Bad keys create hot partitions. A good rule is “same entity → same key; cardinality high enough to spread.”

// Producer: derive a balanced key (e.g., sessionId vs userId)
var props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Optional: custom partitioner when you need domain-aware distribution
props.put("partitioner.class", "com.tbt.CustomHasherPartitioner");

try (var producer = new KafkaProducer<String, String>(props)) {
    String sessionId = SessionKey.derive(userId, deviceId, timestamp); // high-cardinality
    producer.send(new ProducerRecord<>("clicks", sessionId, payload));
}

// Example custom partitioner (simple & fast)
public class CustomHasherPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        var partitions = cluster.partitionsForTopic(topic).size();
        // 64-bit hashing -> positive -> modulo partition count
        long h = java.util.zip.CRC32C.class.desiredAssertionStatus() ? 0 : 0; // placeholder trick to load class
        int hash = java.nio.ByteBuffer.wrap(
            java.security.MessageDigest.getInstance("MD5").digest(keyBytes)
        ).getInt();
        return (hash & 0x7fffffff) % partitions;
    }
    public void close() {}
    public void configure(Map<String, ?> configs) {}
}

Align partitions for joins

If two streams will be joined, both topics must use the same keying and partition count to avoid expensive repartitioning.

# Create aligned topics for a join
kafka-topics.sh --create --topic orders      --partitions 48 --replication-factor 3 --bootstrap-server $BS
kafka-topics.sh --create --topic order-items --partitions 48 --replication-factor 3 --bootstrap-server $BS

// Streams join without shuffle since keys/partitions align
KStream<String, Order> orders = builder.stream("orders");
KStream<String, Item> items = builder.stream("order-items");

orders.join(items,
    (o, i) -> EnrichedOrder.of(o, i),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
).to("enriched-orders");

Safely increase partitions

You can increase partitions, but not decrease. Plan ahead.

# Check current partitions
kafka-topics.sh --describe --topic clicks --bootstrap-server $BS

# Increase partitions during off-peak (producers keep working)
kafka-topics.sh --alter --topic clicks --partitions 96 --bootstrap-server $BS

If consumer lag spikes after increasing, roll out more Streams instances or more threads (see below).

Streams instance and thread model

// Streams config: set threads, state dir, EOS, and internal replication
var props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "clicks-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

// Use multiple threads per instance (match CPU cores sensibly)
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, "4");

// Enable EOS v2 in production
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

// Faster warmup; keep state on SSD-backed path
props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/lib/kstreams/state");

// Internal topics safety
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, "3");
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, "1");

// Caching & commits (tune with real traffic)
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024); // 10MB
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);

Reduce rebalance pain with static membership

// Static membership prevents full rebalances on rolling deploys
props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, System.getenv("POD_NAME")); // unique per pod/VM
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");

In Kubernetes, set POD_NAME via the Downward API so every replica gets a stable instance id.

Handle skew and hot keys

// Split heavy keys into subkeys: "userId#bucket"
KStream<String, Event> events = builder.stream("events");

events
    .selectKey((k, v) -> k + "#" + (v.getTimestamp() / 60_000 % 8)) // 8-way bucketing by minute window
    .groupByKey()
    .count()
    .toStream()
    .to("events-counts");

Downstream, merge buckets when serving results.

Monitor what matters

Quick CLI checks

# Consumer group lag overview
kafka-consumer-groups.sh --bootstrap-server $BS --group clicks-app --describe

# Partition distribution sanity
kafka-topics.sh --describe --topic clicks --bootstrap-server $BS | grep Leader

# JVM heap & open files (on each Streams host)
jcmd $(pgrep -f KafkaStreams) VM.native_memory summary
lsof -p $(pgrep -f KafkaStreams) | wc -l

JMX → Prometheus → Grafana

# jmx_exporter config (jmx_exporter.yaml)
startDelaySeconds: 0
lowercaseOutputName: true
rules:
  - pattern: "kafka.streams<type=stream-metrics, thread-id=(.+), task-id=(.+), topic=(.+)><>([^:]+)"
  - pattern: "kafka.consumer<type=consumer-fetch-manager-metrics, client-id=(.+)><>(records-lag-max|records-consumed-total)"
  - pattern: "rocksdb<name=(.+)><>(block-cache-.*|num-immutable-mem-table|cur-size-active-mem-table|estimate-table-readers-mem)"

# Prometheus scrape snippet
scrape_configs:
  - job_name: 'kstreams'
    metrics_path: /metrics
    static_configs:
      - targets: ['kstreams-0:9404','kstreams-1:9404','kstreams-2:9404']

Useful Grafana panels
• Consumer lag (max, p95) per group and topic
• Records in/out rate
• Rebalance count per hour
• RocksDB size, compaction pauses, block cache hit rate

Back-pressure and commit strategy

// Poll cadence and commit interval influence lag/duplication tradeoffs
props.put(StreamsConfig.POLL_MS_CONFIG, "100"); // tighter polling reduces bursty lag
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "1000"); // 1s commit for fast recovery

If you see steady lag growth, scale threads/instances and verify your sinks (DBs/HTTP) aren’t slow. Offload slow I/O to async buffers.

Tuning state stores

// Materialize with explicit store config
Materialized<String, Long, KeyValueStore<Bytes, byte[]>> store =
    Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store")
        .withCachingEnabled()     // batch writes
        .withLoggingEnabled(Map.of("retention.ms", "86400000")); // 1 day changelog

KStream<String, Event> s = builder.stream("events");
s.groupByKey().count(store).toStream().to("events-counts");

Tip: put the state.dir on SSD; alert on disk > 70%.

Rolling upgrades without drama

# 1) Enable static membership (shown above)
# 2) Roll one instance at a time (Kubernetes maxUnavailable=0)
# 3) Verify group stability
kafka-consumer-groups.sh --bootstrap-server $BS --group clicks-app --describe | grep "STABLE"

Sample topology tying it together

var builder = new StreamsBuilder();

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

// Normalize key, guard against nulls
KStream<String, Click> normalized =
    clicks.selectKey((k, v) -> (v.getSessionId() == null ? "unknown" : v.getSessionId()));

KTable<Windowed<String>, Long> counts =
    normalized
        .groupByKey()
        .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
        .count(Materialized.as("clicks-5m"));

// Export counts
counts.toStream().map((wk, c) -> KeyValue.pair(wk.key(), c)).to("clicks-5m-counts");

// Side-effect sink (e.g., Redis) for serving
counts.toStream().foreach((wk, c) -> RedisClient.set("c5m:" + wk.key(), String.valueOf(c)));

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

Capacity heuristics that work

• Max parallelism ≈ total input partitions
• num.stream.threads ≈ CPU cores per pod minus 1
• Start with cache 10–64MB, commit 1s, EOS v2 on
• Partition count = peak QPS ÷ per-thread sustainable QPS (add 20–30% headroom)

Troubleshooting quick wins

• Lag grows only on a few partitions → key skew; change keying or add bucketing
• Frequent rebalances → set static membership; check flappy pods/Brokers
• RocksDB stalls → reduce block cache pressure, move state to SSD, watch compactions
• EOS throughput low → ensure idempotent producers and fast sinks; consider at-least-once for non-critical paths

Frequently Asked Questions

What’s the fastest way to scale a lagging app?

Increase instances first if partitions ≥ instances. If not, increase partitions, then add instances and threads.

Do I always need exactly-once semantics?

Use EOS for money, inventory, and compliance-critical paths. For analytics/log pipelines, at-least-once is simpler and faster.

How many partitions should I start with?

Enough to cover peak QPS with headroom. If unsure, start at 24–48 for medium workloads and scale up using --alter.

How do I prevent slow sinks from causing lag?

Buffer with an internal queue and push to sinks asynchronously or via a separate worker pool; never block the stream thread on external I/O.

How do I join two topics at scale?

Make sure both use the same key and same partition count; otherwise Streams creates repartition topics and you pay in latency and cost.

5 Views

Top News

Follow Us