You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "peterzdhuang (via GitHub)" <gi...@apache.org> on 2026/07/14 20:48:27 UTC
[PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
peterzdhuang opened a new pull request, #22831:
URL: https://github.com/apache/kafka/pull/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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
Posted by "peterzdhuang (via GitHub)" <gi...@apache.org>.
peterzdhuang closed pull request #22831: KAFKA-19979: Improve Kafka Streams config files
URL: https://github.com/apache/kafka/pull/22831
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
Posted by "ezhou413 (via GitHub)" <gi...@apache.org>.
ezhou413 commented on code in PR #22831:
URL: https://github.com/apache/kafka/pull/22831#discussion_r3631243520
##########
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
+
+# 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), streams.
Review Comment:
`consumer` should also be a value here
##########
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
+
+# 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), 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 (see StreamsConfig for the
+# full list). This avoids conflicts when the same config name means different
+# things to different clients:
Review Comment:
`StreamsConfig` has getter methods for these prefix strings, instead of listing out the actual strings, could we either just point to `StreamsConfig` for more info or mention the getters (to avoid drift)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
Posted by "peterzdhuang (via GitHub)" <gi...@apache.org>.
peterzdhuang commented on PR #22831:
URL: https://github.com/apache/kafka/pull/22831#issuecomment-5039101023
Thanks @ezhou413! Pushed a commit addressing all four:
1. Client-config note: underlying producer/consumer/admin configs can be set here
2. `StreamsConfig` prefixes that scope a config to a client subset
3. Warmup configs: `max.warmup.replicas`, `probing.rebalance.interval.ms`, `acceptable.recovery.lag`
4. `group.protocol`
Ready for another look.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #22831:
URL: https://github.com/apache/kafka/pull/22831#discussion_r4018935774
##########
config/streams.properties:
##########
@@ -0,0 +1,179 @@
+# 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
Review Comment:
I think we should add an entry for `enable.transactional.statestores` right under here with a brief description of what it is - note that enabling this conig only makes sense when the processing guarantee is `exactly_once_v2`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #22831:
URL: https://github.com/apache/kafka/pull/22831#issuecomment-6151119987
Merged #22831 into trunk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck merged PR #22831:
URL: https://github.com/apache/kafka/pull/22831
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19979: Improve Kafka Streams config files [kafka]
Posted by "peterzdhuang (via GitHub)" <gi...@apache.org>.
peterzdhuang commented on code in PR #22831:
URL: https://github.com/apache/kafka/pull/22831#discussion_r4251970447
##########
config/streams.properties:
##########
@@ -0,0 +1,179 @@
+# 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
Review Comment:
Take a look at my newest pr
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org