Configuring Kafka Streams
1. Setting Application ID
| Property | Default | Description |
|---|---|---|
| application.id | null |
The unique identifier for a Kafka Streams application. This ID is used to group streams processors together and manage their state. |
Example: Configuring Application ID
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-processing-app");
2. Configuring Bootstrap Servers
| Property | Default Value | Description |
|---|---|---|
| bootstrap.servers | localhost:9092 | Comma-separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster |
Example: Setting Bootstrap Servers in Kafka Streams Configuration
Properties props = new Properties();
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093");
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streaming-app");Note: Ensure that all brokers in the bootstrap list are reachable from the Kafka Streams application.
3. Setting Default Serdes
| Property | Default Value | Description |
|---|---|---|
| key.serde | null |
Serde used for deserializing keys in streams. If null, the default serde is used. |
| value.serde | null |
Serde used for deserializing values in streams. If null, the default serde is used. |
Example: Configuring Default Serdes
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-app");
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();4. Configuring Commit Interval
| Property | Default Value | Description |
|---|---|---|
| commit.interval.ms | 3000 |
Commit interval in milliseconds for Kafka Streams |
Example: Setting Commit Interval
Properties props = new Properties();
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 5000);
Note: Setting a lower commit interval can improve consistency but may increase latency.
5. Setting Cache Size
| Property | Default Value | Description |
|---|---|---|
| cache.max.bytes.buffering | 1048576 |
Maximum number of bytes to buffer per partition. This setting controls the amount of memory used for caching data in Kafka Streams. |
| cache.max.bytes.buffering | 1048576 |
Maximum number of bytes to buffer per partition. This setting controls the amount of memory used for caching data in Kafka Streams. |
Example: Configuring Cache Size
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 2097152); // 2MB
6. Configuring State Directory
| Property | Default Value | Description |
|---|---|---|
| state.dir | /tmp/kafka-streams |
Directory to store state stores and changelog topics |
| processing.guarantee | at_least_once |
Processing guarantee level for streams (at_least_once, exactly_once_v2) |
Example: Setting State Directory
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.STATE_DIR_CONFIG, "/var/lib/kafka-streams/state");
Note: The state directory must be writable and have sufficient disk space for the application's state stores.
7. Setting Processing Guarantee
| Property | Values | Description |
|---|---|---|
| processing.guarantee | "at_least_once" | "exactly_once_v1" | "exactly_once_v2" |
Defines the processing guarantee level for Kafka Streams. at_least_once is default and ensures messages are processed at least once. exactly_once_v1 provides exactly-once semantics with idempotent producers. exactly_once_v2 offers stronger exactly-once guarantees with transactional processing. |
Example: Configuring Processing Guarantee
Properties props = new Properties();
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
Note:
exactly_once_v2 requires enable.idempotence=true in producer configuration.8. Configuring Replication Factor
| Property | Default Value | Description |
|---|---|---|
| replication.factor | 1 |
Controls the number of replicas for each partition in a Kafka topic. Higher values increase fault tolerance but also increase storage overhead. |
| min.insync.replicas | 1 |
Minimum number of replicas that must acknowledge a write before it is considered successful. Ensures data durability. |
Example: Setting Replication Factor in Topic Creation
kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092
Note: The replication factor must be less than or equal to the number of brokers in the cluster.
9. Setting Number of Threads
| Property | Default Value | Description |
|---|---|---|
| num.threads | 1 |
Number of threads to use for processing records in the Kafka Streams application. |
Example: Configuring Number of Threads
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.NUM_THREADS_CONFIG, 3); // Set to 3 threads
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();Note: Increasing the number of threads allows for parallel processing of records, which can improve throughput in high-volume scenarios.
10. Configuring Window Store Retention
Window stores in Kafka Streams manage time-based data aggregations. Properly configuring window store retention is crucial for balancing data availability with storage efficiency.
| Property | Default Value | Description |
|---|---|---|
| window.store.retention.ms | 24 hours (86400000 ms) |
Duration for which windowed data is retained before being purged. |
| window.store.retention.ms | 24 hours (86400000 ms) |
Duration for which windowed data is retained before being purged. |
Example: Setting Window Store Retention
StreamsConfig config = new StreamsConfig();
config.put(StreamsConfig.WINDOW_STORE_RETENTION_MS_CONFIG, 172800000L); // 48 hours
Note: The retention period must be at least as long as the window size to ensure data is not prematurely removed during aggregation.