Configuring Kafka Streams

1. Setting Application ID

Config Description Detail
application.id Unique app identifier = consumer group
Prefix Internal topic names app-id-...
Stable Don't change in prod Resets state

Example: Application id

props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");

2. Configuring Bootstrap Servers

Config Description Detail
bootstrap.servers Cluster entry points Required
Multiple List for resilience Comma-separated
Discovery Full cluster learned Auto

Example: Bootstrap servers

props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "b1:9092,b2:9092");

3. Setting Default Serdes

Config Description Detail
DEFAULT_KEY_SERDE Key serializer/deserializer Class
DEFAULT_VALUE_SERDE Value serde Class
Override Per-operator Consumed/Produced Local

Example: Default serdes

props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
  Serdes.String().getClass());

4. Configuring Commit Interval

Config Description Default
commit.interval.ms State commit frequency 30000 / 100 (eos)
Lower Less reprocessing on crash More overhead
EOS Defaults to 100ms Tighter

Example: Commit interval

props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10000);

5. Setting Cache Size

Config Description Default
statestore.cache.
max.bytes		 Record cache for dedup NEW 10MB
0 Emit every update No dedup
Larger Fewer downstream emits More memory

Example: Cache size

props.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 52428800);

6. Configuring State Directory

Config Description Default
state.dir Local state location /tmp/kafka-streams
Persistent Use durable disk in prod Not /tmp
Per-Instance Unique per app instance No sharing

Example: State dir

props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/lib/streams");

7. Setting Processing Guarantee

Value Description Detail
at_least_once Default, may dup Fast
exactly_once_v2 EOS via transactions Recommended
Requirement RF ≥ 3, min ISR 2 Durability

Example: Exactly-once

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
  StreamsConfig.EXACTLY_ONCE_V2);

8. Configuring Replication Factor

Config Description Prod
replication.factor Internal topic RF 3
Applies To Changelog + repartition All internal
-1 Use broker default 3.x+

Example: Internal topic RF

props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);

9. Setting Number of Threads

Config Description Default
num.stream.threads Processing threads 1
Parallelism Bounded by partitions ≤ tasks
Scaling Add threads or instances Rebalances

Example: Stream threads

props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);

10. Configuring Window Store Retention

Aspect Description Detail
Materialized.
withRetention		 Window store retain time ≥ window + grace
windowstore.
changelog.additional.
retention.ms		 Changelog buffer Default 1d
Expiry Old windows dropped Bounded state

Example: Window retention

Materialized.as("win-store")
  .withRetention(Duration.ofHours(1));

Related Cheat Sheets

Kafka

Using KRaft Consensus Protocol

Kafka

Monitoring Broker Metrics

Kafka

Optimizing Broker Performance

Kafka

Implementing Custom Partitioners