You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/08/27 04:38:43 UTC
[GitHub] [kafka] ableegoldman opened a new pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
ableegoldman opened a new pull request #11272:
URL: https://github.com/apache/kafka/pull/11272
Most configs that are read and used by Streams today originate from the properties passed in to the KafkaStreams constructor, which means they get applied universally across all threads, tasks, subtopologies, and so on. The only current exception to this is the `topology.optimization` config which is parsed from the properties that get passed in to `StreamsBuilder#build`. However there are a handful of configs that could also be scoped to the topology level, allowing users to configure each NamedTopology independently of the others, where it makes sense to do so.
This PR refactors the handling of these configs by interpreting the values passed in via KafkaStreams constructor as the global defaults, which can then be overridden for individual topologies via the properties passed in when building the NamedTopology. More topology-level configs may be added in the future, but this PR covers the following:
1. `max.task.idle.ms`
2. `task.timeout.ms`
3. `buffered.records.per.partition`
4. `default.timestamp.extractor.class`
5. `default.deserialization.exception.handler`
--
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
[GitHub] [kafka] wcarlson5 commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
wcarlson5 commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r697495900
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
##########
@@ -67,7 +68,7 @@
stateDirectory,
stateMgr,
inputPartitions,
- config.getLong(StreamsConfig.TASK_TIMEOUT_MS_CONFIG),
Review comment:
Do we want to have the same pattern for task level configs as we do for streams configs?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java
##########
@@ -262,7 +262,7 @@ private StreamTask createActiveTask(final TaskId taskId,
inputPartitions,
topology,
consumer,
- config,
+ topologyMetadata.getTaskConfigFor(taskId),
Review comment:
Maybe we take config as well for any overrides? I don't know what would take presidents. Or maybe check for conflicts?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -359,15 +362,24 @@ public final InternalTopologyBuilder setApplicationId(final String applicationId
return this;
}
- public synchronized final InternalTopologyBuilder setStreamsConfig(final StreamsConfig config) {
- Objects.requireNonNull(config, "config can't be null");
- this.config = config;
+ public synchronized final void setTopologyProperties(final Properties props) {
+ this.topologyProperties = props;
+ }
- return this;
+ public synchronized final void setStreamsConfig(final StreamsConfig config) {
+ Objects.requireNonNull(config, "config can't be null");
+ this.applicationConfig = config;
+ topologyConfigs = new TopologyConfig(
+ topologyName,
+ applicationConfig,
+ topologyProperties == null ?
Review comment:
Do we use Optional in streams? this would be a good place for it
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -133,7 +135,9 @@
private Map<Integer, Set<String>> nodeGroups = null;
- private StreamsConfig config = null;
+ private StreamsConfig applicationConfig = null; // the global streams configs and default topology props
+ private Properties topologyProperties = null; // this topology's config overrides
Review comment:
I am not clear on the difference based on the names can we just name it `topologyOverrides`?
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704887980
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -359,15 +362,24 @@ public final InternalTopologyBuilder setApplicationId(final String applicationId
return this;
}
- public synchronized final InternalTopologyBuilder setStreamsConfig(final StreamsConfig config) {
- Objects.requireNonNull(config, "config can't be null");
- this.config = config;
+ public synchronized final void setTopologyProperties(final Properties props) {
+ this.topologyProperties = props;
+ }
- return this;
+ public synchronized final void setStreamsConfig(final StreamsConfig config) {
+ Objects.requireNonNull(config, "config can't be null");
+ this.applicationConfig = config;
+ topologyConfigs = new TopologyConfig(
+ topologyName,
+ applicationConfig,
+ topologyProperties == null ?
Review comment:
Technically Optional is not supposed to be used for fields, only return values...so they say. I'm definitely a little concerned about this pattern of null fields but it's already used throughout `InternalTopologyBuilder` and often unavoidable due to the particular order of things in Streams. So yet another thing to clean up at a later time (think guozhang filed a ticket for this already)
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704916241
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -611,6 +611,7 @@ public synchronized Topology build() {
*/
public synchronized Topology build(final Properties props) {
internalStreamsBuilder.buildAndOptimizeTopology(props);
+ internalTopologyBuilder.setTopologyProperties(props);
Review comment:
To clarify, these are the props that the user passes in when building the topology, eg needed for `TOPOLOGY_OPTIMIZATION` which is the only actual topology-level override today. So the overrides are only set the same as the global props if the user decides to pass the same set of configs in.
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704906629
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
+ .define(MAX_TASK_IDLE_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ MAX_TASK_IDLE_MS_DOC)
+ .define(TASK_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ TASK_TIMEOUT_MS_DOC);
+ }
+ private final Logger log = LoggerFactory.getLogger(TopologyConfig.class);
+
+ public final String topologyName;
+ public final boolean eosEnabled;
+
+ final long maxTaskIdleMs;
+ final long taskTimeoutMs;
+ final int maxBufferedSize;
+ final Supplier<TimestampExtractor> timestampExtractorSupplier;
+ final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+
+ public TopologyConfig(final String topologyName, final StreamsConfig globalAppConfigs, final Properties topologyOverrides) {
+ super(CONFIG, topologyOverrides, false);
+
+ this.topologyName = topologyName;
+ this.eosEnabled = StreamThread.eosEnabled(globalAppConfigs);
+
+ if (isTopologyOverride(MAX_TASK_IDLE_MS_CONFIG, topologyOverrides)) {
+ maxTaskIdleMs = getLong(MAX_TASK_IDLE_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, MAX_TASK_IDLE_MS_CONFIG, maxTaskIdleMs);
+ } else {
+ maxTaskIdleMs = globalAppConfigs.getLong(MAX_TASK_IDLE_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(TASK_TIMEOUT_MS_CONFIG, topologyOverrides)) {
+ taskTimeoutMs = getLong(TASK_TIMEOUT_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, TASK_TIMEOUT_MS_CONFIG, taskTimeoutMs);
+ } else {
+ taskTimeoutMs = globalAppConfigs.getLong(TASK_TIMEOUT_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
+ maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
+ } else {
+ maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+
+ if (isTopologyOverride(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, topologyOverrides)) {
+ timestampExtractorSupplier = () -> getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, getClass(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG));
+ } else {
+ timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ }
+
+ if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
+ deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ } else {
+ deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+
+ }
+ }
+
+ /**
+ * @return true if there is an override for this config in the properties of this NamedTopology. Applications that
+ * don't use named topologies will just refer to the global defaults regardless of the topology properties
+ */
+ private boolean isTopologyOverride(final String config, final Properties topologyOverrides) {
+ return topologyName != null && topologyOverrides.containsKey(config);
Review comment:
Added a TODO for now
--
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
[GitHub] [kafka] guozhangwang commented on pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#issuecomment-919603159
`Checkstyle rule violations were found.` :)
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r706092654
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -359,15 +362,24 @@ public final InternalTopologyBuilder setApplicationId(final String applicationId
return this;
}
- public synchronized final InternalTopologyBuilder setStreamsConfig(final StreamsConfig config) {
- Objects.requireNonNull(config, "config can't be null");
- this.config = config;
+ public synchronized final void setTopologyProperties(final Properties props) {
+ this.topologyProperties = props;
+ }
- return this;
+ public synchronized final void setStreamsConfig(final StreamsConfig config) {
Review comment:
@guozhangwang Played around with this a bit and was able to clean things up nicely, plus address some other awkwardness that was bugging me. Worked out great (just need to do some cleanup of the tests now...)
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r711723716
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java
##########
@@ -112,10 +113,42 @@ private KafkaStreamsNamedTopologyWrapper(final Collection<NamedTopology> topolog
);
}
+ /**
+ * Provides a high-level DSL for specifying the processing logic of your application and building it into an
+ * independent topology that can be executed by this {@link KafkaStreams}.
+ *
+ * @param topologyName The name for this topology
+ * @param topologyConfigs The properties and any config overrides for this topology
+ *
+ * @throws IllegalArgumentException if the name contains the character sequence "__"
+ */
+ public NamedTopologyBuilder newNamedTopologyBuilder(final String topologyName, final Properties topologyConfigs) {
+ if (topologyName.contains(TaskId.NAMED_TOPOLOGY_DELIMITER)) {
+ throw new IllegalArgumentException("The character sequence '__' is not allowed in a NamedTopology, please select a new name");
+ }
+ return new NamedTopologyBuilder(topologyName, applicationConfigs, topologyConfigs);
+ }
+
+ /**
+ * Returns an empty topology for full control over the graph of streams and processor nodes that define the processing
+ * logic to be executed by this {@link KafkaStreams}.
+ *
+ * @param topologyName The name for this topology
+ * @param topologyConfigs The properties and any config overrides for this topology
+ *
+ * @throws IllegalArgumentException if the name contains the character sequence "__"
+ */
+ public NamedTopology newNamedTopology(final String topologyName, final Properties topologyConfigs) {
Review comment:
Well eventually we'll want to support building a NamedTopology with the PAPI rather than the DSL, but I can remove this for the time being to keep the API minimal until we've figured out exactly what it will look like & gone through the KIP
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r706014148
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -359,15 +362,24 @@ public final InternalTopologyBuilder setApplicationId(final String applicationId
return this;
}
- public synchronized final InternalTopologyBuilder setStreamsConfig(final StreamsConfig config) {
- Objects.requireNonNull(config, "config can't be null");
- this.config = config;
+ public synchronized final void setTopologyProperties(final Properties props) {
+ this.topologyProperties = props;
+ }
- return this;
+ public synchronized final void setStreamsConfig(final StreamsConfig config) {
Review comment:
That all makes sense to me -- initially I was trying to make this just work with the existing public API, but ultimately figured that even the concept of "topology-level overrides" would require a KIP. If we're only dealing with named topologies then we're free to do things the right way from the start -- thanks for the suggestions :)
--
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
[GitHub] [kafka] ableegoldman merged pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman merged pull request #11272:
URL: https://github.com/apache/kafka/pull/11272
--
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
[GitHub] [kafka] guozhangwang commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r708747814
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -611,6 +611,7 @@ public synchronized Topology build() {
*/
public synchronized Topology build(final Properties props) {
internalStreamsBuilder.buildAndOptimizeTopology(props);
+ internalTopologyBuilder.setTopologyProperties(props);
Review comment:
Yeah something like https://github.com/apache/kafka/pull/11272/files#diff-0e5e608831150c058e2ad1b45d38ad941739562588ec0fdb97cc9f742919fb1fR145
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704890418
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
+ .define(MAX_TASK_IDLE_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ MAX_TASK_IDLE_MS_DOC)
+ .define(TASK_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ TASK_TIMEOUT_MS_DOC);
+ }
+ private final Logger log = LoggerFactory.getLogger(TopologyConfig.class);
+
+ public final String topologyName;
+ public final boolean eosEnabled;
+
+ final long maxTaskIdleMs;
+ final long taskTimeoutMs;
+ final int maxBufferedSize;
+ final Supplier<TimestampExtractor> timestampExtractorSupplier;
+ final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+
+ public TopologyConfig(final String topologyName, final StreamsConfig globalAppConfigs, final Properties topologyOverrides) {
+ super(CONFIG, topologyOverrides, false);
+
+ this.topologyName = topologyName;
+ this.eosEnabled = StreamThread.eosEnabled(globalAppConfigs);
+
+ if (isTopologyOverride(MAX_TASK_IDLE_MS_CONFIG, topologyOverrides)) {
+ maxTaskIdleMs = getLong(MAX_TASK_IDLE_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, MAX_TASK_IDLE_MS_CONFIG, maxTaskIdleMs);
+ } else {
+ maxTaskIdleMs = globalAppConfigs.getLong(MAX_TASK_IDLE_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(TASK_TIMEOUT_MS_CONFIG, topologyOverrides)) {
+ taskTimeoutMs = getLong(TASK_TIMEOUT_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, TASK_TIMEOUT_MS_CONFIG, taskTimeoutMs);
+ } else {
+ taskTimeoutMs = globalAppConfigs.getLong(TASK_TIMEOUT_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
+ maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
+ } else {
+ maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+
+ if (isTopologyOverride(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, topologyOverrides)) {
+ timestampExtractorSupplier = () -> getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, getClass(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG));
+ } else {
+ timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ }
+
+ if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
+ deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ } else {
+ deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+
+ }
+ }
+
+ /**
+ * @return true if there is an override for this config in the properties of this NamedTopology. Applications that
+ * don't use named topologies will just refer to the global defaults regardless of the topology properties
+ */
+ private boolean isTopologyOverride(final String config, final Properties topologyOverrides) {
+ return topologyName != null && topologyOverrides.containsKey(config);
Review comment:
Yeah, this NamedTopology stuff is still an experimental feature and hasn't gone through a KIP so we shouldn't expect the topology name to be set anyways (yet)
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704907727
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java
##########
@@ -262,7 +262,7 @@ private StreamTask createActiveTask(final TaskId taskId,
inputPartitions,
topology,
consumer,
- config,
+ topologyMetadata.getTaskConfigFor(taskId),
Review comment:
Can you clarify? Not sure if this is what you meant, but this should return the actual task configs, with any overrides already applied. I tried to keep all the override & config resolution logic inside the TopologyConfig class
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704888258
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review comment:
where's checkstyle when you need it 🙄
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r699942903
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review comment:
indent is wrong here.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
Review comment:
Should we mention that currently we have 5 topology level properties: BUFFERED_RECORDS_PER_PARTITION_CONFIG..., and this list should be updated if new topology level config is introduced in the future?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
+ .define(MAX_TASK_IDLE_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ MAX_TASK_IDLE_MS_DOC)
+ .define(TASK_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ TASK_TIMEOUT_MS_DOC);
+ }
+ private final Logger log = LoggerFactory.getLogger(TopologyConfig.class);
+
+ public final String topologyName;
+ public final boolean eosEnabled;
+
+ final long maxTaskIdleMs;
+ final long taskTimeoutMs;
+ final int maxBufferedSize;
+ final Supplier<TimestampExtractor> timestampExtractorSupplier;
+ final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+
+ public TopologyConfig(final String topologyName, final StreamsConfig globalAppConfigs, final Properties topologyOverrides) {
+ super(CONFIG, topologyOverrides, false);
+
+ this.topologyName = topologyName;
+ this.eosEnabled = StreamThread.eosEnabled(globalAppConfigs);
+
+ if (isTopologyOverride(MAX_TASK_IDLE_MS_CONFIG, topologyOverrides)) {
+ maxTaskIdleMs = getLong(MAX_TASK_IDLE_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, MAX_TASK_IDLE_MS_CONFIG, maxTaskIdleMs);
+ } else {
+ maxTaskIdleMs = globalAppConfigs.getLong(MAX_TASK_IDLE_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(TASK_TIMEOUT_MS_CONFIG, topologyOverrides)) {
+ taskTimeoutMs = getLong(TASK_TIMEOUT_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, TASK_TIMEOUT_MS_CONFIG, taskTimeoutMs);
+ } else {
+ taskTimeoutMs = globalAppConfigs.getLong(TASK_TIMEOUT_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
+ maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
+ } else {
+ maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+
+ if (isTopologyOverride(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, topologyOverrides)) {
+ timestampExtractorSupplier = () -> getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, getClass(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG));
+ } else {
+ timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ }
+
+ if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
+ deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ } else {
+ deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+
+ }
+ }
+
+ /**
+ * @return true if there is an override for this config in the properties of this NamedTopology. Applications that
+ * don't use named topologies will just refer to the global defaults regardless of the topology properties
+ */
+ private boolean isTopologyOverride(final String config, final Properties topologyOverrides) {
+ return topologyName != null && topologyOverrides.containsKey(config);
Review comment:
Should we have warnning log for users to let them know their properties doesn't get overridden because of no topology name set? Otherwise, the user will never know it, right?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
+ .define(MAX_TASK_IDLE_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ MAX_TASK_IDLE_MS_DOC)
+ .define(TASK_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ TASK_TIMEOUT_MS_DOC);
+ }
+ private final Logger log = LoggerFactory.getLogger(TopologyConfig.class);
+
+ public final String topologyName;
+ public final boolean eosEnabled;
+
+ final long maxTaskIdleMs;
+ final long taskTimeoutMs;
+ final int maxBufferedSize;
+ final Supplier<TimestampExtractor> timestampExtractorSupplier;
+ final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+
+ public TopologyConfig(final String topologyName, final StreamsConfig globalAppConfigs, final Properties topologyOverrides) {
+ super(CONFIG, topologyOverrides, false);
+
+ this.topologyName = topologyName;
+ this.eosEnabled = StreamThread.eosEnabled(globalAppConfigs);
+
+ if (isTopologyOverride(MAX_TASK_IDLE_MS_CONFIG, topologyOverrides)) {
+ maxTaskIdleMs = getLong(MAX_TASK_IDLE_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, MAX_TASK_IDLE_MS_CONFIG, maxTaskIdleMs);
+ } else {
+ maxTaskIdleMs = globalAppConfigs.getLong(MAX_TASK_IDLE_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(TASK_TIMEOUT_MS_CONFIG, topologyOverrides)) {
+ taskTimeoutMs = getLong(TASK_TIMEOUT_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, TASK_TIMEOUT_MS_CONFIG, taskTimeoutMs);
+ } else {
+ taskTimeoutMs = globalAppConfigs.getLong(TASK_TIMEOUT_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
+ maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
+ } else {
+ maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+
+ if (isTopologyOverride(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, topologyOverrides)) {
+ timestampExtractorSupplier = () -> getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, getClass(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG));
+ } else {
+ timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ }
+
+ if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
+ deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ } else {
+ deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+
Review comment:
additional new line.
--
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
[GitHub] [kafka] showuon commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r699942903
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
Review comment:
indent is wrong here.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
Review comment:
Should we mention that currently we have 5 topology level properties: BUFFERED_RECORDS_PER_PARTITION_CONFIG..., and this list should be updated if new topology level config is introduced in the future?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
+ .define(MAX_TASK_IDLE_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ MAX_TASK_IDLE_MS_DOC)
+ .define(TASK_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ TASK_TIMEOUT_MS_DOC);
+ }
+ private final Logger log = LoggerFactory.getLogger(TopologyConfig.class);
+
+ public final String topologyName;
+ public final boolean eosEnabled;
+
+ final long maxTaskIdleMs;
+ final long taskTimeoutMs;
+ final int maxBufferedSize;
+ final Supplier<TimestampExtractor> timestampExtractorSupplier;
+ final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+
+ public TopologyConfig(final String topologyName, final StreamsConfig globalAppConfigs, final Properties topologyOverrides) {
+ super(CONFIG, topologyOverrides, false);
+
+ this.topologyName = topologyName;
+ this.eosEnabled = StreamThread.eosEnabled(globalAppConfigs);
+
+ if (isTopologyOverride(MAX_TASK_IDLE_MS_CONFIG, topologyOverrides)) {
+ maxTaskIdleMs = getLong(MAX_TASK_IDLE_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, MAX_TASK_IDLE_MS_CONFIG, maxTaskIdleMs);
+ } else {
+ maxTaskIdleMs = globalAppConfigs.getLong(MAX_TASK_IDLE_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(TASK_TIMEOUT_MS_CONFIG, topologyOverrides)) {
+ taskTimeoutMs = getLong(TASK_TIMEOUT_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, TASK_TIMEOUT_MS_CONFIG, taskTimeoutMs);
+ } else {
+ taskTimeoutMs = globalAppConfigs.getLong(TASK_TIMEOUT_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
+ maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
+ } else {
+ maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+
+ if (isTopologyOverride(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, topologyOverrides)) {
+ timestampExtractorSupplier = () -> getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, getClass(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG));
+ } else {
+ timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ }
+
+ if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
+ deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ } else {
+ deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+
+ }
+ }
+
+ /**
+ * @return true if there is an override for this config in the properties of this NamedTopology. Applications that
+ * don't use named topologies will just refer to the global defaults regardless of the topology properties
+ */
+ private boolean isTopologyOverride(final String config, final Properties topologyOverrides) {
+ return topologyName != null && topologyOverrides.containsKey(config);
Review comment:
Should we have warnning log for users to let them know their properties doesn't get overridden because of no topology name set? Otherwise, the user will never know it, right?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
+ .define(MAX_TASK_IDLE_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ MAX_TASK_IDLE_MS_DOC)
+ .define(TASK_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ TASK_TIMEOUT_MS_DOC);
+ }
+ private final Logger log = LoggerFactory.getLogger(TopologyConfig.class);
+
+ public final String topologyName;
+ public final boolean eosEnabled;
+
+ final long maxTaskIdleMs;
+ final long taskTimeoutMs;
+ final int maxBufferedSize;
+ final Supplier<TimestampExtractor> timestampExtractorSupplier;
+ final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+
+ public TopologyConfig(final String topologyName, final StreamsConfig globalAppConfigs, final Properties topologyOverrides) {
+ super(CONFIG, topologyOverrides, false);
+
+ this.topologyName = topologyName;
+ this.eosEnabled = StreamThread.eosEnabled(globalAppConfigs);
+
+ if (isTopologyOverride(MAX_TASK_IDLE_MS_CONFIG, topologyOverrides)) {
+ maxTaskIdleMs = getLong(MAX_TASK_IDLE_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, MAX_TASK_IDLE_MS_CONFIG, maxTaskIdleMs);
+ } else {
+ maxTaskIdleMs = globalAppConfigs.getLong(MAX_TASK_IDLE_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(TASK_TIMEOUT_MS_CONFIG, topologyOverrides)) {
+ taskTimeoutMs = getLong(TASK_TIMEOUT_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, TASK_TIMEOUT_MS_CONFIG, taskTimeoutMs);
+ } else {
+ taskTimeoutMs = globalAppConfigs.getLong(TASK_TIMEOUT_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
+ maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
+ } else {
+ maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+
+ if (isTopologyOverride(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, topologyOverrides)) {
+ timestampExtractorSupplier = () -> getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, getClass(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG));
+ } else {
+ timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ }
+
+ if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
+ deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ } else {
+ deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+
Review comment:
additional new line.
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704907727
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java
##########
@@ -262,7 +262,7 @@ private StreamTask createActiveTask(final TaskId taskId,
inputPartitions,
topology,
consumer,
- config,
+ topologyMetadata.getTaskConfigFor(taskId),
Review comment:
Sorry can you clarify? This should return the actual task configs, with any overrides already applied. I tried to keep all the override & config resolution logic inside the TopologyConfig class
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704886569
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java
##########
@@ -67,7 +68,7 @@
stateDirectory,
stateMgr,
inputPartitions,
- config.getLong(StreamsConfig.TASK_TIMEOUT_MS_CONFIG),
Review comment:
Do you mean like a separate `TaskConfig` public class for users to configure the same way we do for `StreamsConfig`? I've been going back and forth but ultimately I think we should keep all configs under StreamsConfig so that all configs for each part of the stack are handled in one place (eg it would be confusing if within the Consumer client, there were multiple sets of configs)...still I think we can clean this up eventually, when we do the KIP.
--
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
[GitHub] [kafka] guozhangwang commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r708751965
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java
##########
@@ -112,10 +113,42 @@ private KafkaStreamsNamedTopologyWrapper(final Collection<NamedTopology> topolog
);
}
+ /**
+ * Provides a high-level DSL for specifying the processing logic of your application and building it into an
+ * independent topology that can be executed by this {@link KafkaStreams}.
+ *
+ * @param topologyName The name for this topology
+ * @param topologyConfigs The properties and any config overrides for this topology
+ *
+ * @throws IllegalArgumentException if the name contains the character sequence "__"
+ */
+ public NamedTopologyBuilder newNamedTopologyBuilder(final String topologyName, final Properties topologyConfigs) {
+ if (topologyName.contains(TaskId.NAMED_TOPOLOGY_DELIMITER)) {
+ throw new IllegalArgumentException("The character sequence '__' is not allowed in a NamedTopology, please select a new name");
+ }
+ return new NamedTopologyBuilder(topologyName, applicationConfigs, topologyConfigs);
+ }
+
+ /**
+ * Returns an empty topology for full control over the graph of streams and processor nodes that define the processing
+ * logic to be executed by this {@link KafkaStreams}.
+ *
+ * @param topologyName The name for this topology
+ * @param topologyConfigs The properties and any config overrides for this topology
+ *
+ * @throws IllegalArgumentException if the name contains the character sequence "__"
+ */
+ public NamedTopology newNamedTopology(final String topologyName, final Properties topologyConfigs) {
Review comment:
This one seems not used? And even in external callers like ksql, I think we would go through the newNamedTopologyBuilder().build() right?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -359,15 +363,23 @@ public final InternalTopologyBuilder setApplicationId(final String applicationId
return this;
}
- public synchronized final InternalTopologyBuilder setStreamsConfig(final StreamsConfig config) {
- Objects.requireNonNull(config, "config can't be null");
- this.config = config;
+ public synchronized final void setTopologyOverrides(final Properties props) {
Review comment:
This function seems not used in non-testing code any more?
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java
##########
@@ -927,10 +931,54 @@ public void shouldSetCorrectSourceNodesWithRegexUpdatedTopics() {
}
@Test
- public void shouldSetStreamsConfigOnRewriteTopology() {
+ public void shouldSetTopologyConfigOnRewriteTopology() {
+ final Properties globalProps = StreamsTestUtils.getStreamsConfig();
+ globalProps.put(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, 100L);
+ final StreamsConfig globalStreamsConfig = new StreamsConfig(globalProps);
+ final InternalTopologyBuilder topologyBuilder = builder.rewriteTopology(globalStreamsConfig);
+ assertThat(topologyBuilder.topologyConfig(), equalTo(new TopologyConfig(null, globalStreamsConfig, new Properties())));
+ assertThat(topologyBuilder.topologyConfig().getTaskConfig().maxTaskIdleMs, equalTo(100L));
+ }
+
+ @Test
+ public void shouldOverrideGlobalStreamsConfigWhenGivenNamedTopologyProps() {
+ final Properties topologyOverrides = new Properties();
+ topologyOverrides.put(StreamsConfig.MAX_TASK_IDLE_MS_CONFIG, 500L);
+ topologyOverrides.put(StreamsConfig.TASK_TIMEOUT_MS_CONFIG, 1000L);
+ topologyOverrides.put(StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG, 15);
+ topologyOverrides.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, MockTimestampExtractor.class);
+ topologyOverrides.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class);
+ builder.setNamedTopology(new NamedTopology("test-topology", new Properties()));
+ builder.setTopologyOverrides(topologyOverrides);
+
+ final StreamsConfig config = new StreamsConfig(StreamsTestUtils.getStreamsConfig());
+ final InternalTopologyBuilder topologyBuilder = builder.rewriteTopology(config);
+
+ assertThat(topologyBuilder.topologyConfig().getTaskConfig().maxTaskIdleMs, equalTo(500L));
+ assertThat(topologyBuilder.topologyConfig().getTaskConfig().taskTimeoutMs, equalTo(1000L));
+ assertThat(topologyBuilder.topologyConfig().getTaskConfig().maxBufferedSize, equalTo(15));
+ assertThat(topologyBuilder.topologyConfig().getTaskConfig().timestampExtractor.getClass(), equalTo(MockTimestampExtractor.class));
+ assertThat(topologyBuilder.topologyConfig().getTaskConfig().deserializationExceptionHandler.getClass(), equalTo(LogAndContinueExceptionHandler.class));
+ }
+
+ @Test
+ public void shouldNotOverrideGlobalStreamsConfigWhenGivenUnnamedTopologyProps() {
Review comment:
Thanks for the updated tests! They are great.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java
##########
@@ -112,10 +113,42 @@ private KafkaStreamsNamedTopologyWrapper(final Collection<NamedTopology> topolog
);
}
+ /**
+ * Provides a high-level DSL for specifying the processing logic of your application and building it into an
+ * independent topology that can be executed by this {@link KafkaStreams}.
+ *
+ * @param topologyName The name for this topology
+ * @param topologyConfigs The properties and any config overrides for this topology
+ *
+ * @throws IllegalArgumentException if the name contains the character sequence "__"
+ */
+ public NamedTopologyBuilder newNamedTopologyBuilder(final String topologyName, final Properties topologyConfigs) {
+ if (topologyName.contains(TaskId.NAMED_TOPOLOGY_DELIMITER)) {
+ throw new IllegalArgumentException("The character sequence '__' is not allowed in a NamedTopology, please select a new name");
+ }
+ return new NamedTopologyBuilder(topologyName, applicationConfigs, topologyConfigs);
+ }
+
+ /**
+ * Returns an empty topology for full control over the graph of streams and processor nodes that define the processing
+ * logic to be executed by this {@link KafkaStreams}.
+ *
+ * @param topologyName The name for this topology
+ * @param topologyConfigs The properties and any config overrides for this topology
+ *
+ * @throws IllegalArgumentException if the name contains the character sequence "__"
+ */
+ public NamedTopology newNamedTopology(final String topologyName, final Properties topologyConfigs) {
Review comment:
nit: in some classes we call it `topologyConfigs` and in some others we call it `topologyOverrides` (e.g. the NamedTopology constructor which takes `topologyConfigs` but name it as `topologyOverrides`). How about just make all of these parameters `topologyConfigs`?
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704909244
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
Review comment:
I guess, but honestly for now it's going to be on us/me to keep track of any new configs that could be topology overrides. It's not a problem if someone introduces a config that could be scoped to a single topology and isn't, it's just a feature we can expand later. We need to work out a clean API before taking this to the KIP stage anyways
--
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
[GitHub] [kafka] ableegoldman commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r704909650
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -611,6 +611,7 @@ public synchronized Topology build() {
*/
public synchronized Topology build(final Properties props) {
internalStreamsBuilder.buildAndOptimizeTopology(props);
+ internalTopologyBuilder.setTopologyProperties(props);
Review comment:
Like this https://github.com/apache/kafka/pull/11272/files#diff-0e5e608831150c058e2ad1b45d38ad941739562588ec0fdb97cc9f742919fb1fR139 ? Or were you referring to something else
--
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
[GitHub] [kafka] ableegoldman commented on pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#issuecomment-906922473
@wcarlson5 split the config work out from the Pt. 4 branch, here's the PR as promised. I ended up refactoring things a bit while I was pulling the changes out to make things easier to build on top of as we go. Let me know what you think, especially any thoughts on something we can do now to better set up the next piece: `cache.max.bytes.buffering`,
We should also keep an eye on KIP-770 and start thinking about the `input.buffer.max.bytes` config it's proposing. Might be good for you to be one of the primary reviewers, if you have time, so you can guide things to play along nicely with our work here.
--
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
[GitHub] [kafka] guozhangwang commented on a change in pull request #11272: KAFKA-12648: introduce TopologyConfig and TaskConfig for topology-level overrides
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11272:
URL: https://github.com/apache/kafka/pull/11272#discussion_r700479414
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -611,6 +611,7 @@ public synchronized Topology build() {
*/
public synchronized Topology build(final Properties props) {
internalStreamsBuilder.buildAndOptimizeTopology(props);
+ internalTopologyBuilder.setTopologyProperties(props);
Review comment:
nit: add a TODO here that for now we always set the overrides as the same as global application props?
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/NamedTopologyStreamsBuilder.java
##########
@@ -17,35 +17,24 @@
package org.apache.kafka.streams.processor.internals.namedtopology;
import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.processor.TaskId;
import java.util.Properties;
public class NamedTopologyStreamsBuilder extends StreamsBuilder {
- final String topologyName;
-
/**
* @param topologyName any string representing your NamedTopology, all characters allowed except for "__"
* @throws IllegalArgumentException if the name contains the character sequence "__"
*/
public NamedTopologyStreamsBuilder(final String topologyName) {
- super();
- this.topologyName = topologyName;
+ super(new NamedTopology(topologyName));
Review comment:
Nice cleanup!
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -133,7 +135,9 @@
private Map<Integer, Set<String>> nodeGroups = null;
- private StreamsConfig config = null;
+ private StreamsConfig applicationConfig = null; // the global streams configs and default topology props
+ private Properties topologyProperties = null; // this topology's config overrides
Review comment:
I also feel these three var names are a bit confusing. `topologyOverrides` reads better.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java
##########
@@ -101,7 +101,7 @@ void removeRevokedUnknownTasks(final Set<TaskId> assignedTasks) {
final ProcessorStateManager stateManager = new ProcessorStateManager(
taskId,
Task.TaskType.STANDBY,
- StreamThread.eosEnabled(config),
+ StreamThread.eosEnabled(applicationConfig),
Review comment:
This is not for this PR, but originally the function `StreamThread.eosEnabled` is introduced in the `StreamThread` class as a quick-hack since adding it to `StreamsConfigs` would be a public API and requires a KIP. It is used in 1) streams-producer, 2) tasks, 3) in tests. Now that we have removed the usage in 2) with TaskConfigs (which is an internal class). I feel in the long run 1)'s use-case may also be replaced by another internal class, say ApplicationConfig or even TopologyConfig (if we can manage to have separate values of this within a single app in the future), and at that time we should migrate this function out of `StreamThread`.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.streams.processor.internals.namedtopology;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+import org.apache.kafka.streams.processor.internals.StreamThread;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.Properties;
+import java.util.function.Supplier;
+
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.BUFFERED_RECORDS_PER_PARTITION_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.TASK_TIMEOUT_MS_DOC;
+
+/**
+ * Streams configs that apply at the topology level. The values in the {@link StreamsConfig} parameter of the
+ * {@link org.apache.kafka.streams.KafkaStreams} or {@link KafkaStreamsNamedTopologyWrapper} constructors will
+ * determine the defaults, which can then be overridden for specific topologies by passing them in when creating the
+ * topology via the {@link org.apache.kafka.streams.StreamsBuilder#build(Properties)} or
+ * {@link NamedTopologyStreamsBuilder#buildNamedTopology(Properties)} methods.
+ */
+public class TopologyConfig extends AbstractConfig {
+ private static final ConfigDef CONFIG;
+ static {
+ CONFIG = new ConfigDef()
+ .define(BUFFERED_RECORDS_PER_PARTITION_CONFIG,
+ Type.INT,
+ null,
+ Importance.LOW,
+ BUFFERED_RECORDS_PER_PARTITION_DOC)
+ .define(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_DOC)
+ .define(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
+ Type.CLASS,
+ null,
+ Importance.MEDIUM,
+ DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_DOC)
+ .define(MAX_TASK_IDLE_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ MAX_TASK_IDLE_MS_DOC)
+ .define(TASK_TIMEOUT_MS_CONFIG,
+ Type.LONG,
+ null,
+ Importance.MEDIUM,
+ TASK_TIMEOUT_MS_DOC);
+ }
+ private final Logger log = LoggerFactory.getLogger(TopologyConfig.class);
+
+ public final String topologyName;
+ public final boolean eosEnabled;
+
+ final long maxTaskIdleMs;
+ final long taskTimeoutMs;
+ final int maxBufferedSize;
+ final Supplier<TimestampExtractor> timestampExtractorSupplier;
+ final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+
+ public TopologyConfig(final String topologyName, final StreamsConfig globalAppConfigs, final Properties topologyOverrides) {
+ super(CONFIG, topologyOverrides, false);
+
+ this.topologyName = topologyName;
+ this.eosEnabled = StreamThread.eosEnabled(globalAppConfigs);
+
+ if (isTopologyOverride(MAX_TASK_IDLE_MS_CONFIG, topologyOverrides)) {
+ maxTaskIdleMs = getLong(MAX_TASK_IDLE_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, MAX_TASK_IDLE_MS_CONFIG, maxTaskIdleMs);
+ } else {
+ maxTaskIdleMs = globalAppConfigs.getLong(MAX_TASK_IDLE_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(TASK_TIMEOUT_MS_CONFIG, topologyOverrides)) {
+ taskTimeoutMs = getLong(TASK_TIMEOUT_MS_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, TASK_TIMEOUT_MS_CONFIG, taskTimeoutMs);
+ } else {
+ taskTimeoutMs = globalAppConfigs.getLong(TASK_TIMEOUT_MS_CONFIG);
+ }
+
+ if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
+ maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, BUFFERED_RECORDS_PER_PARTITION_CONFIG, maxBufferedSize);
+ } else {
+ maxBufferedSize = globalAppConfigs.getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
+ }
+
+ if (isTopologyOverride(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, topologyOverrides)) {
+ timestampExtractorSupplier = () -> getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, getClass(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG));
+ } else {
+ timestampExtractorSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TimestampExtractor.class);
+ }
+
+ if (isTopologyOverride(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, topologyOverrides)) {
+ deserializationExceptionHandlerSupplier = () -> getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, getClass(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG));
+ } else {
+ deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
+
+ }
+ }
+
+ /**
+ * @return true if there is an override for this config in the properties of this NamedTopology. Applications that
+ * don't use named topologies will just refer to the global defaults regardless of the topology properties
+ */
+ private boolean isTopologyOverride(final String config, final Properties topologyOverrides) {
+ return topologyName != null && topologyOverrides.containsKey(config);
Review comment:
I think `topologyName == null` only when there's no named topology, which could still be quite common as people are not yet aware of the new APIs yet. In this case logging an warn may be surprising to them.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
##########
@@ -359,15 +362,24 @@ public final InternalTopologyBuilder setApplicationId(final String applicationId
return this;
}
- public synchronized final InternalTopologyBuilder setStreamsConfig(final StreamsConfig config) {
- Objects.requireNonNull(config, "config can't be null");
- this.config = config;
+ public synchronized final void setTopologyProperties(final Properties props) {
+ this.topologyProperties = props;
+ }
- return this;
+ public synchronized final void setStreamsConfig(final StreamsConfig config) {
Review comment:
This is a meta comment: I feel we can try to re-order the topology steps in a better way now. Today we have the following in order which looks awkward to me.
1) InternalStreamsBuilder#internalTopologyBuilder(props): this is to do the logical plan -> physical plan build. The `props` should be a per-topology overrides in the long run.
2) InternalStreamsBuilder#rewriteTopology(StreamsConfig config): this is to do some post-plan generation modifications which, ideally, should be part of 1) as well. Here the `config` is at global application-level.
3) Inside 2) we call `setStreamsConfig(config);` which is to set the global application configs.
I think a better order should be first setting the global configs, and then the per-topology overrides. So we should consider setting the global configs at the beginning when constructing the Topology (either from StreamsBuilder with DSL, or directly with PAPI), then the per-topology props for overrides.
Of course the reason we have this awkwardness is because for PAPI, the Topology is built without "props" passed in at all and hence we can only rely on the last step 3) to make sure the configs are finally initialized. But now with the new `NamedTopology` we can bypass this by requiring it to always come with a prop at the construction time as well, which would be used to set both (the func names are just for illustrations):
* internalTopologyBuilder.setTopologyOverrides(props);
* internalTopologyBuilder.setApplicationConfigs(config);
We would end up with the following scenarios:
1. No named-topology, DSL: props are set at StreamsBuilder.build(), and we know there's no overrides.
2. No named-topology, PAPI: the only pain left, as we have to wait at InternalTopologyBuilder.rewriteTopology, but still we know there's no overrides.
3. Named-topology, DSL: props are set at constructor, for both application-level and overrides.
4. Named-topology, PAPI: currently no one should be using in this way, but just in case it is supported it is still the same as 3) above.
And if in the long run we would stick with the NamedTopology where we would deprecate 1/2 above, we would end up naturally at the state where we enforce props to be set at the topology construction time.
WDYT?
--
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