(kafka) branch trunk updated: KAFKA-19979: Improve Kafka Streams config files (#22831)

[bb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 4d4cc7c414b KAFKA-19979: Improve Kafka Streams config files (#22831)
4d4cc7c414b is described below

commit 4d4cc7c414b0a23ee08806d62a464c84eae7250b
Author: Peter Huang <13...@users.noreply.github.com>
AuthorDate: Tue Oct 13 10:17:02 2026 -0700

    KAFKA-19979: Improve Kafka Streams config files (#22831)
    
    Adds a config/streams.properties template with sections and comments,
    similar to the other properties files shipped with Kafka. Same idea as
    KAFKA-19574, which did this for the producer and consumer configs.
    
    Based on #21308 by @mshijin-ksolves, which stalled after review, and
    picks up @ezhou413's review comments from there. Changes from that PR:
    
    - Commented out `default.key.serde` / `default.value.serde` (they have
    no
      default) and listed valid values instead
    - Removed the `<>` around `bootstrap.servers`
    - Noted that `commit.interval.ms` defaults to 100 under
    `exactly_once_v2`
      (otherwise 30000)
    - Used `statestore.cache.max.bytes` instead of the deprecated
      `cache.max.bytes.buffering`
    - Listed the valid `topology.optimization` values
    - Fixed the serde class name to `Serdes$StringSerde`
    
    Also adds a test that validates the file against StreamsConfig so it
    can't drift out of sync.
    
    Reviewers: Evan Zhou <ez...@confluent.io>, Bill Bejeck
     <bb...@apache.org>
---
 config/streams.properties                          | 184 +++++++++++++++++++++
 .../apache/kafka/streams/StreamsConfigTest.java    |  24 +++
 2 files changed, 208 insertions(+)

diff --git a/config/streams.properties b/config/streams.properties
new file mode 100644
index 00000000000..71b33710f9a
--- /dev/null
+++ b/config/streams.properties
@@ -0,0 +1,184 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# See org.apache.kafka.streams.StreamsConfig for more details.
+# Consider using environment variables or external configuration management
+# for sensitive information like passwords and environment-specific settings.
+
+##################### Application Basics #####################
+
+# A unique identifier for the Kafka Streams application.
+# Used as the consumer group id and as a prefix for internal (changelog and
+# repartition) topic names, so it must be unique within the cluster.
+# REQUIRED (no default).
+application.id=my-streams-app
+
+# List of Kafka brokers used for initial cluster discovery and metadata retrieval.
+# Format: host1:port1,host2:port2,host3:port3
+# REQUIRED (no default).
+bootstrap.servers=localhost:9092
+
+##################### Stream Threads #####################
+
+# Number of threads to execute stream processing.
+# Default: 1
+num.stream.threads=1
+
+##################### Default SerDes #####################
+
+# Default serializer/deserializer for record keys and values. These configs
+# have no default, so a SerDe must either be set here or supplied explicitly in
+# the code (for example via Consumed, Produced, or Materialized). A SerDe passed
+# in the code always takes precedence over the default set here.
+# Possible values include the SerDes in org.apache.kafka.common.serialization,
+# for example:
+#   org.apache.kafka.common.serialization.Serdes$StringSerde
+#   org.apache.kafka.common.serialization.Serdes$ByteArraySerde
+#   org.apache.kafka.common.serialization.Serdes$LongSerde
+#default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
+#default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
+
+##################### Processing Guarantees #####################
+
+# The processing guarantee that should be used.
+# Options: at_least_once (default), exactly_once_v2.
+processing.guarantee=at_least_once
+
+# Buffer state store writes until the corresponding changelog transaction
+# commits. Enable only with processing.guarantee=exactly_once_v2.
+# Default: false
+enable.transactional.statestores=false
+
+# The frequency in milliseconds with which to commit processing progress.
+# Note: the default depends on processing.guarantee. When processing.guarantee
+# is exactly_once_v2 the default is 100, otherwise the default is 30000.
+commit.interval.ms=30000
+
+##################### State Store #####################
+
+# Directory location for local state stores.
+# Default: ${java.io.tmpdir}/kafka-streams
+state.dir=/tmp/kafka-streams
+
+##################### Task Assignment #####################
+
+# The number of standby replicas for each task. Standby replicas keep shadow
+# copies of local state so a task can be resumed on another instance without
+# rebuilding its state from the changelog.
+# Default: 0
+num.standby.replicas=0
+
+# The maximum number of warmup replicas (extra standbys beyond num.standby.replicas)
+# that can be assigned at once. Warmup replicas let a task catch up its state on a
+# new instance in the background before it takes over as active, so moving tasks
+# does not stall processing.
+# Default: 2
+max.warmup.replicas=2
+
+# The maximum time in milliseconds to wait before triggering a rebalance to probe
+# for warmup replicas that have finished warming up and are ready to become active.
+# Default: 600000 (10 minutes)
+probing.rebalance.interval.ms=600000
+
+# The maximum acceptable lag (number of offsets behind) for a task's state to be
+# considered caught up enough to receive an active task assignment. A task lagging
+# by more than this is only assigned as a warmup/standby until it catches up.
+# Default: 10000
+acceptable.recovery.lag=10000
+
+# The group protocol Kafka Streams should use.
+# Options: classic (default), consumer, streams.
+# With "streams" the broker-side Streams rebalance protocol is used and task
+# assignment is managed by the broker. In that mode the client-side warmup and
+# standby settings above (max.warmup.replicas, num.standby.replicas) are ignored
+# (a warning is logged); configure standby replicas broker-side instead.
+group.protocol=classic
+
+##################### Caching #####################
+
+# Maximum number of memory bytes to be used for state store caching across all
+# threads. Replaces the deprecated cache.max.bytes.buffering.
+# Default: 10485760 (10 MB)
+statestore.cache.max.bytes=10485760
+
+##################### Internal Topics #####################
+
+# Replication factor for changelog and repartition topics created by the
+# application.
+# Default: -1 (use the broker default)
+replication.factor=-1
+
+##################### Topology Optimization #####################
+
+# A comma-separated list of optimizations applied to the topology, or one of the
+# aggregate values.
+# Options: none (default), all, or any combination of
+# reuse.ktable.source.topics, merge.repartition.topics,
+# single.store.self.join.
+topology.optimization=none
+
+##################### Metrics #####################
+
+# The window of time (ms) a metrics sample is computed over.
+# Default: 30000
+metrics.sample.window.ms=30000
+
+# The number of samples maintained to compute metrics.
+# Default: 2
+metrics.num.samples=2
+
+# The highest recording level for metrics.
+# Options: INFO (default), DEBUG, TRACE.
+metrics.recording.level=INFO
+
+##################### Timestamp Extraction #####################
+
+# Default timestamp extractor class that implements the
+# org.apache.kafka.streams.processor.TimestampExtractor interface.
+# Options include: FailOnInvalidTimestamp (default), LogAndSkipOnInvalidTimestamp,
+# UsePartitionTimeOnInvalidTimestamp, WallclockTimestampExtractor.
+default.timestamp.extractor=org.apache.kafka.streams.processor.FailOnInvalidTimestamp
+
+##################### Client Configuration Overrides #####################
+
+# Kafka Streams runs on top of the regular Kafka producer, consumer, and admin
+# clients, so any of their configs can be set in this file to tune the underlying
+# clients. A config with no prefix is passed to every client that accepts it.
+#
+# To target a single client type, prefix the config. This avoids conflicts when
+# the same config name means different things to different clients. See
+# org.apache.kafka.streams.StreamsConfig for the full set of prefixes, the
+# clients each one applies to, and the precedence between them.
+#
+# Examples (commented out):
+#producer.acks=all
+#consumer.max.poll.records=1000
+#main.consumer.fetch.max.bytes=1048576
+
+##################### Security Configuration #####################
+
+# Security protocol used to communicate with brokers.
+# Options: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
+#security.protocol=SASL_SSL
+
+# SSL configuration.
+#ssl.truststore.location=/path/to/truststore.jks
+#ssl.truststore.password=truststore-password
+
+# SASL configuration.
+#sasl.mechanism=PLAIN
+#sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+#    username="your-username" \
+#    password="your-password";
diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
index c9cefc97218..edd9217b267 100644
--- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java
@@ -53,6 +53,8 @@ import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.File;
+import java.io.FileInputStream;
+import java.io.InputStream;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 import java.nio.charset.StandardCharsets;
@@ -2193,6 +2195,28 @@ public class StreamsConfigTest {
         assertTrue(streamsConfig.getBoolean(TRANSACTIONAL_STATE_STORES_CONFIG));
     }
 
+    /**
+     * Validates config/streams.properties file to avoid getting out of sync with StreamsConfig.
+     */
+    @Test
+    public void testValidateConfigPropertiesFile() {
+        final Properties fileProps = new Properties();
+
+        try (InputStream inputStream = new FileInputStream(System.getProperty("user.dir") + "/../config/streams.properties")) {
+            fileProps.load(inputStream);
+        } catch (final Exception e) {
+            fail("Failed to load config/streams.properties file: " + e.getMessage());
+        }
+
+        final StreamsConfig config = new StreamsConfig(fileProps);
+
+        for (final String key : config.originals().keySet()) {
+            if (!StreamsConfig.configDef().configKeys().containsKey(key)) {
+                fail("Invalid configuration key: " + key);
+            }
+        }
+    }
+
     static class MisconfiguredSerde implements Serde<Object> {
         @Override
         public void configure(final Map<String, ?>  configs, final boolean isKey) {