You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sebastienviale (via GitHub)" <gi...@apache.org> on 2024/12/17 15:51:21 UTC
[PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
sebastienviale opened a new pull request, #18233:
URL: https://github.com/apache/kafka/pull/18233
Pull request to implement KIP-1111, aims to add a configuration that prevents a Kafka Streams application from starting if any of its internal topics have auto-generated names, thereby enforcing explicit naming for all internal topics and enhancing the stability of the application’s topology.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1948836873
##########
streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java:
##########
@@ -2354,6 +2363,418 @@ public void shouldNowAllowStreamAndTableFromSameTopic() {
assertThrows(TopologyException.class, builder::build);
}
+ @ParameterizedTest
+ @CsvSource({
+ "true, false, true, false",
+ "false, true, true, true",
+ "true, true, true, true",
+ "false, false, false, false"
+ })
+ public void groupByWithAggregationTest(final boolean isGroupByKeyNamed,
+ final boolean isMaterializedNamed,
+ final boolean isLoggingEnabled,
+ final boolean isValid) {
+ final Map<Object, Object> props = dummyStreamsConfigMap();
+ props.put(ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG, true);
+
+ final Grouped<String, String> grouped;
+ final Materialized<String, Long, KeyValueStore<Bytes, byte[]>> materialized;
+ if (isGroupByKeyNamed) {
+ grouped = Grouped.with("repartition-name", Serdes.String(), Serdes.String());
+ } else {
+ grouped = Grouped.with(Serdes.String(), Serdes.String());
+ }
+ if (isMaterializedNamed) {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("materialized-name")
+ .withKeySerde(Serdes.String()).withValueSerde(Serdes.Long());
+ } else {
+ if (isLoggingEnabled) {
+ materialized = Materialized.with(Serdes.String(), Serdes.Long());
+ } else {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>with(Serdes.String(), Serdes.Long())
+ .withLoggingDisabled();
+ }
+ }
+
+ final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));
+ final KStream<String, String> stream = builder.stream("input1");
+ stream
+ .groupBy((k, v) -> v, grouped)
+ .count(materialized)
+ .toStream()
+ .to("output", Produced.as("sink"));
+
+ if (isValid) {
+ assertDoesNotThrow(() -> builder.build());
+ } else {
+ final TopologyException e = assertThrows(TopologyException.class, () -> builder.build());
+ // TODO check the error with loggingDisabled
Review Comment:
all unit tests have been refactored
##########
streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java:
##########
@@ -2354,6 +2363,418 @@ public void shouldNowAllowStreamAndTableFromSameTopic() {
assertThrows(TopologyException.class, builder::build);
}
+ @ParameterizedTest
+ @CsvSource({
+ "true, false, true, false",
+ "false, true, true, true",
+ "true, true, true, true",
+ "false, false, false, false"
+ })
+ public void groupByWithAggregationTest(final boolean isGroupByKeyNamed,
Review Comment:
all unit tests have been refactored
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937427960
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalResourcesNaming.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+public final class InternalResourcesNaming {
+
+ private String repartitionTopic;
+ private String changelogTopic;
+ private String stateStore;
+
+ private InternalResourcesNaming() {
+ }
+
+ public static InternalResourcesNaming build() {
Review Comment:
done, make it a real builder
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalResourcesNaming.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+public final class InternalResourcesNaming {
+
+ private String repartitionTopic;
+ private String changelogTopic;
+ private String stateStore;
+
+ private InternalResourcesNaming() {
+ }
+
+ public static InternalResourcesNaming build() {
Review Comment:
done, made it a real builder
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2659644207
@lucasbru done all the changes except the log message in case the new config is false.
I added some documentation, it may be improved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2674933743
> @sebastienviale Your changes to the docs look good to me. I wonder if it would also make sense to add a little paragraph to docs/streams/developer-guide/dsl-topology-naming.html where it makes sense? Just to point the reader to the new config.
I just added a little paragraph to the conclusion, or do you think it makes more sense to add it before the conclusion ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2675414685
LGTM Could you check why there is a test failing? FAILED ❌ StreamsConfigTest > shouldNotLeakInternalDocMembers() looks related
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1966317130
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,12 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal" +
Review Comment:
doc string should not be public 🙂
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937447061
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2333,4 +2341,33 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addUnprovidedInternalTopics(final InternalResourcesNaming internalResourcesNaming) {
+ unprovidedInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!unprovidedInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ for (final InternalResourcesNaming internalResourcesNaming : unprovidedInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ result.append(String.format("Following changelog topic has not been named: %s%n", internalResourcesNaming.changelogTopic()));
Review Comment:
I think it is a good idea, thanks !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1922600453
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,10 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal topics (e.g., changelog and repartition topics) and their associated state stores." +
+ " When enabled, the application will fail to start if any internal resource has an auto-generated name.";
Review Comment:
```suggestion
" When enabled, the application will refuse to start if any internal resource has an auto-generated name.";
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -869,6 +873,11 @@ public class StreamsConfig extends AbstractConfig {
Importance.HIGH,
STATE_DIR_DOC,
"${java.io.tmpdir}")
+ .define(ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG,
+ Type.BOOLEAN,
Review Comment:
nit: indentation is off
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,10 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal topics (e.g., changelog and repartition topics) and their associated state stores." +
Review Comment:
nit: line wrap?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java:
##########
@@ -214,8 +215,7 @@ private void processRepartitions(final Map<KGroupedStreamImpl<K, ?>, Aggregator<
final String repartitionNamePrefix = repartitionReqs.userProvidedRepartitionTopicName != null ?
repartitionReqs.userProvidedRepartitionTopicName : storeName;
- createRepartitionSource(repartitionNamePrefix, repartitionNodeBuilder, repartitionReqs.keySerde, repartitionReqs.valueSerde);
-
+ createRepartitionSource(repartitionNamePrefix, repartitionNodeBuilder, repartitionReqs.keySerde, repartitionReqs.valueSerde, repartitionReqs.userProvidedRepartitionTopicName != null || queryableName != null);
Review Comment:
nit: Can we store the boolean expression in a local variable and give it a name, so it's easier to understand what you are passing (will also shorten the line).
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -192,6 +196,10 @@ public InternalTopologyBuilder(final TopologyConfig topologyConfigs) {
private boolean hasPersistentStores = false;
+ private boolean ensureExplicitInternalResourceNaming;
+
+ private Set<InternalResourcesNaming> unprovidedInternalNames = new LinkedHashSet<>();
Review Comment:
could be final
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java:
##########
@@ -204,7 +204,8 @@ <KR> KTable<KR, VOut> build(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? sup
}
private void processRepartitions(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? super K, ? super Object, VOut>> groupPatterns,
Review Comment:
nit: if you put the last two arguments on a separate line, also put `groupPatterns` on a separate line.
##########
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java:
##########
@@ -384,7 +384,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materializedInternal =
new MaterializedInternal<>(
- Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()),
+ Materialized.<K, V, KeyValueStore<Bytes, byte[]>>with(consumedInternal.keySerde(), consumedInternal.valueSerde()).withLoggingDisabled(),
Review Comment:
Not sure I understand why you are disabling logging here
##########
streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java:
##########
@@ -2354,6 +2363,418 @@ public void shouldNowAllowStreamAndTableFromSameTopic() {
assertThrows(TopologyException.class, builder::build);
}
+ @ParameterizedTest
+ @CsvSource({
+ "true, false, true, false",
+ "false, true, true, true",
+ "true, true, true, true",
+ "false, false, false, false"
+ })
+ public void groupByWithAggregationTest(final boolean isGroupByKeyNamed,
+ final boolean isMaterializedNamed,
+ final boolean isLoggingEnabled,
+ final boolean isValid) {
+ final Map<Object, Object> props = dummyStreamsConfigMap();
+ props.put(ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG, true);
+
+ final Grouped<String, String> grouped;
+ final Materialized<String, Long, KeyValueStore<Bytes, byte[]>> materialized;
+ if (isGroupByKeyNamed) {
+ grouped = Grouped.with("repartition-name", Serdes.String(), Serdes.String());
+ } else {
+ grouped = Grouped.with(Serdes.String(), Serdes.String());
+ }
+ if (isMaterializedNamed) {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("materialized-name")
+ .withKeySerde(Serdes.String()).withValueSerde(Serdes.Long());
+ } else {
+ if (isLoggingEnabled) {
+ materialized = Materialized.with(Serdes.String(), Serdes.Long());
+ } else {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>with(Serdes.String(), Serdes.Long())
+ .withLoggingDisabled();
+ }
+ }
+
+ final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));
+ final KStream<String, String> stream = builder.stream("input1");
+ stream
+ .groupBy((k, v) -> v, grouped)
+ .count(materialized)
+ .toStream()
+ .to("output", Produced.as("sink"));
+
+ if (isValid) {
+ assertDoesNotThrow(() -> builder.build());
+ } else {
+ final TopologyException e = assertThrows(TopologyException.class, () -> builder.build());
Review Comment:
Can we check the exception message? It seems rather important here. Generally, if this will become hard to implement, maybe we can use 4 separate tests instead of a parametrized test, and a private method to avoid repeating everything 4 times. Same goes for the tests below.
Alternatively, we could check the exception message in a unit test for `InternalTopicBuilder`.
##########
streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java:
##########
@@ -2354,6 +2363,418 @@ public void shouldNowAllowStreamAndTableFromSameTopic() {
assertThrows(TopologyException.class, builder::build);
}
+ @ParameterizedTest
+ @CsvSource({
+ "true, false, true, false",
+ "false, true, true, true",
+ "true, true, true, true",
+ "false, false, false, false"
+ })
+ public void groupByWithAggregationTest(final boolean isGroupByKeyNamed,
+ final boolean isMaterializedNamed,
+ final boolean isLoggingEnabled,
+ final boolean isValid) {
+ final Map<Object, Object> props = dummyStreamsConfigMap();
+ props.put(ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG, true);
+
+ final Grouped<String, String> grouped;
+ final Materialized<String, Long, KeyValueStore<Bytes, byte[]>> materialized;
+ if (isGroupByKeyNamed) {
+ grouped = Grouped.with("repartition-name", Serdes.String(), Serdes.String());
+ } else {
+ grouped = Grouped.with(Serdes.String(), Serdes.String());
+ }
+ if (isMaterializedNamed) {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("materialized-name")
+ .withKeySerde(Serdes.String()).withValueSerde(Serdes.Long());
+ } else {
+ if (isLoggingEnabled) {
+ materialized = Materialized.with(Serdes.String(), Serdes.Long());
+ } else {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>with(Serdes.String(), Serdes.Long())
+ .withLoggingDisabled();
+ }
+ }
+
+ final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));
+ final KStream<String, String> stream = builder.stream("input1");
+ stream
+ .groupBy((k, v) -> v, grouped)
+ .count(materialized)
+ .toStream()
+ .to("output", Produced.as("sink"));
+
+ if (isValid) {
+ assertDoesNotThrow(() -> builder.build());
+ } else {
+ final TopologyException e = assertThrows(TopologyException.class, () -> builder.build());
+ // TODO check the error with loggingDisabled
Review Comment:
Do you plan to address this `TODO` in this PR?
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -489,7 +492,8 @@ private void mergeRepartitionTopics() {
//passing in the name of the first repartition topic, re-used to create the optimized repartition topic
final GraphNode optimizedSingleRepartition = createRepartitionNode(repartitionTopicName,
groupedInternal.keySerde(),
- groupedInternal.valueSerde());
+ groupedInternal.valueSerde(),
+ true);
Review Comment:
It seems we can remove the new boolean parameter here, since it's always true. It's also a bit misleading, because the repartition topic name isn't necessarily provided by the user, we just don't check it at this point.
##########
streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java:
##########
@@ -2354,6 +2363,418 @@ public void shouldNowAllowStreamAndTableFromSameTopic() {
assertThrows(TopologyException.class, builder::build);
}
+ @ParameterizedTest
+ @CsvSource({
+ "true, false, true, false",
+ "false, true, true, true",
+ "true, true, true, true",
+ "false, false, false, false"
+ })
+ public void groupByWithAggregationTest(final boolean isGroupByKeyNamed,
Review Comment:
The name of the test should probably follow the `should...` format as the other tests, and mention something about detected implicit internal resource names. The same goes for the other tests below.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2333,4 +2341,33 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addUnprovidedInternalTopics(final InternalResourcesNaming internalResourcesNaming) {
Review Comment:
Replace "Topics" by "Names" as you are also talking about state stores?
Additionally, I was wondering if you have considered using something like `implicit` instead of `unprovided` as it's clearer that it's the opposite of `explicit` in the config?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java:
##########
@@ -457,7 +457,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
Objects.requireNonNull(materialized, "materialized can't be null");
final ConsumedInternal<K, V> consumedInternal = new ConsumedInternal<>(consumed);
// always use the serdes from consumed
- materialized.withKeySerde(consumedInternal.keySerde()).withValueSerde(consumedInternal.valueSerde());
+ materialized.withKeySerde(consumedInternal.keySerde()).withValueSerde(consumedInternal.valueSerde()).withLoggingDisabled();
Review Comment:
Same as above
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2333,4 +2341,33 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addUnprovidedInternalTopics(final InternalResourcesNaming internalResourcesNaming) {
+ unprovidedInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!unprovidedInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ for (final InternalResourcesNaming internalResourcesNaming : unprovidedInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ result.append(String.format("Following changelog topic has not been named: %s%n", internalResourcesNaming.changelogTopic()));
Review Comment:
It could be nicer to collect the resource names by type and output one line per type? As in:
```
Following changelog topic has not been named: a, b, c
```
This is just a suggestion
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -192,6 +196,10 @@ public InternalTopologyBuilder(final TopologyConfig topologyConfigs) {
private boolean hasPersistentStores = false;
+ private boolean ensureExplicitInternalResourceNaming;
Review Comment:
Could be final
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalResourcesNaming.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+public final class InternalResourcesNaming {
+
+ private String repartitionTopic;
+ private String changelogTopic;
+ private String stateStore;
+
+ private InternalResourcesNaming() {
+ }
+
+ public static InternalResourcesNaming build() {
Review Comment:
This class at first looks like a builder, but it's something different, since you build first and change values afterwards. I would add a proper builder, or just use a regular class with a public constructor and setters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1956770352
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
I don't necessarily have a problem with the warning log in theory, but I think we need to be careful to specify that you shouldn't add names if they have been missing from the start. This warning is really only useful if people see it before deploying their app the first time.
I guess it might still help them remember to add names the next time they write an app, but we should definitely note this in the log. But the current wording seems to suggest they should stop and add names to everything which will mess up an app that already has data in the changelogs/repartitions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937324922
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java:
##########
@@ -214,8 +215,7 @@ private void processRepartitions(final Map<KGroupedStreamImpl<K, ?>, Aggregator<
final String repartitionNamePrefix = repartitionReqs.userProvidedRepartitionTopicName != null ?
repartitionReqs.userProvidedRepartitionTopicName : storeName;
- createRepartitionSource(repartitionNamePrefix, repartitionNodeBuilder, repartitionReqs.keySerde, repartitionReqs.valueSerde);
-
+ createRepartitionSource(repartitionNamePrefix, repartitionNodeBuilder, repartitionReqs.keySerde, repartitionReqs.valueSerde, repartitionReqs.userProvidedRepartitionTopicName != null || queryableName != null);
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2591478643
Just one high level comment: can we add the necessary docs update to this PR (ie, adding the new config to the config section, and adding a new `4.1` upgrade section and list the KIP there -- not sure if there is anything else we would need?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1958184392
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
So since nobody else seems to be too worried about the warning and the KIP was accepted as it is, we can keep the warning. But Sophie raises a good point; make sure to add a note that changing internal resource names will typically require an application reset to keep a long story short. wdyt, @ableegoldman?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937350055
##########
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java:
##########
@@ -384,7 +384,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materializedInternal =
new MaterializedInternal<>(
- Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()),
+ Materialized.<K, V, KeyValueStore<Bytes, byte[]>>with(consumedInternal.keySerde(), consumedInternal.valueSerde()).withLoggingDisabled(),
Review Comment:
For a GlobalKTable logging is always disable, I have to specify it here otherwise a changelog topic will be added as a not named internal resource:
https://github.com/sebastienviale/kafka/blob/a8790a02ff1f1de7d387b1cdaf9b93b349e6403e/streams/src/main/java/org/apache/kafka/streams/kstream/internals/MaterializedInternal.java#L54-L64
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937384414
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -192,6 +196,10 @@ public InternalTopologyBuilder(final TopologyConfig topologyConfigs) {
private boolean hasPersistentStores = false;
+ private boolean ensureExplicitInternalResourceNaming;
Review Comment:
done
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -192,6 +196,10 @@ public InternalTopologyBuilder(final TopologyConfig topologyConfigs) {
private boolean hasPersistentStores = false;
+ private boolean ensureExplicitInternalResourceNaming;
+
+ private Set<InternalResourcesNaming> unprovidedInternalNames = new LinkedHashSet<>();
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937325997
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java:
##########
@@ -489,7 +492,8 @@ private void mergeRepartitionTopics() {
//passing in the name of the first repartition topic, re-used to create the optimized repartition topic
final GraphNode optimizedSingleRepartition = createRepartitionNode(repartitionTopicName,
groupedInternal.keySerde(),
- groupedInternal.valueSerde());
+ groupedInternal.valueSerde(),
+ true);
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1966316686
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1948837239
##########
streams/src/test/java/org/apache/kafka/streams/StreamsBuilderTest.java:
##########
@@ -2354,6 +2363,418 @@ public void shouldNowAllowStreamAndTableFromSameTopic() {
assertThrows(TopologyException.class, builder::build);
}
+ @ParameterizedTest
+ @CsvSource({
+ "true, false, true, false",
+ "false, true, true, true",
+ "true, true, true, true",
+ "false, false, false, false"
+ })
+ public void groupByWithAggregationTest(final boolean isGroupByKeyNamed,
+ final boolean isMaterializedNamed,
+ final boolean isLoggingEnabled,
+ final boolean isValid) {
+ final Map<Object, Object> props = dummyStreamsConfigMap();
+ props.put(ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG, true);
+
+ final Grouped<String, String> grouped;
+ final Materialized<String, Long, KeyValueStore<Bytes, byte[]>> materialized;
+ if (isGroupByKeyNamed) {
+ grouped = Grouped.with("repartition-name", Serdes.String(), Serdes.String());
+ } else {
+ grouped = Grouped.with(Serdes.String(), Serdes.String());
+ }
+ if (isMaterializedNamed) {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("materialized-name")
+ .withKeySerde(Serdes.String()).withValueSerde(Serdes.Long());
+ } else {
+ if (isLoggingEnabled) {
+ materialized = Materialized.with(Serdes.String(), Serdes.Long());
+ } else {
+ materialized = Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>with(Serdes.String(), Serdes.Long())
+ .withLoggingDisabled();
+ }
+ }
+
+ final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(props)));
+ final KStream<String, String> stream = builder.stream("input1");
+ stream
+ .groupBy((k, v) -> v, grouped)
+ .count(materialized)
+ .toStream()
+ .to("output", Produced.as("sink"));
+
+ if (isValid) {
+ assertDoesNotThrow(() -> builder.build());
+ } else {
+ final TopologyException e = assertThrows(TopologyException.class, () -> builder.build());
Review Comment:
all unit tests have been refactored
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937323668
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,10 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal topics (e.g., changelog and repartition topics) and their associated state stores." +
Review Comment:
done
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -869,6 +873,11 @@ public class StreamsConfig extends AbstractConfig {
Importance.HIGH,
STATE_DIR_DOC,
"${java.io.tmpdir}")
+ .define(ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG,
+ Type.BOOLEAN,
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1965533413
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
Explicit naming for internal resources is currently disabled. If you want to enforce user-defined names for all internal resources, set "ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG" to true. Note: Changing internal resource names may require a full streams application reset for an already deployed application. Consult the documentation on naming operators for more details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1967136311
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,12 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal" +
Review Comment:
I had to change it to default access modifier (package-private) because TopologyConfig needs 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
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2677621132
> LGTM Could you check why there is a test failing? FAILED ❌ StreamsConfigTest > shouldNotLeakInternalDocMembers() looks related
Regarding @ableegoldman comment I had to add the ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC in the exception list to make the test succeed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937351094
##########
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java:
##########
@@ -457,7 +457,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
Objects.requireNonNull(materialized, "materialized can't be null");
final ConsumedInternal<K, V> consumedInternal = new ConsumedInternal<>(consumed);
// always use the serdes from consumed
- materialized.withKeySerde(consumedInternal.keySerde()).withValueSerde(consumedInternal.valueSerde());
+ materialized.withKeySerde(consumedInternal.keySerde()).withValueSerde(consumedInternal.valueSerde()).withLoggingDisabled();
Review Comment:
Same as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1952422436
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImplJoin.java:
##########
@@ -168,6 +169,15 @@ public <K, V1, V2, VOut> KStream<K, VOut> join(final KStream<K, V1> lhs,
);
}
+ if (userProvidedBaseStoreName == null) {
+ addInternalResourceName(thisWindowStore);
+ addInternalResourceName(otherWindowStore);
+ if (outerJoinWindowStore.isPresent()) {
+ addInternalResourceName(outerJoinWindowStore.get());
+ }
+
+ }
Review Comment:
```suggestion
}
}
```
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java:
##########
@@ -1051,76 +1063,15 @@ public <VTable, VOut> KStream<K, VOut> leftJoin(final KTable<K, VTable> table,
final KStreamImpl<K, V> thisStreamRepartitioned = repartitionForJoin(
name != null ? name : this.name,
joinedInternal.keySerde(),
- joinedInternal.leftValueSerde()
+ joinedInternal.leftValueSerde(),
+ name != null
);
return thisStreamRepartitioned.doStreamTableJoin(table, joiner, joinedInternal, true);
} else {
return doStreamTableJoin(table, joiner, joinedInternal, true);
}
}
- @SuppressWarnings({"unchecked", "resource"})
Review Comment:
Why did you move this method? It seems Matthias moved it on purpose and you are reverting his change here?
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,11 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal " +
+ " topics (e.g., changelog and repartition topics) and their associated state stores." +
+ " When enabled, the application will refuse to start if any internal resource has an auto-generated name.";
/** {@code log.summary.interval.ms} */
Review Comment:
nit: missing newline above javadoc
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,11 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal " +
Review Comment:
```suggestion
public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal" +
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
I'm not sure I'm mega happy with this log message, as every user trying a basic example of Kafka Streams will be confronted with it. But I see it was part of the KIP and this proposal was accepted. I added a little note to the VOTE thread, to point out that this log was part of the KIP - I had missed it. But I'm okay with going forward with the KIP as accepted if nobody shares my concerns.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -64,12 +66,15 @@
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import static org.apache.kafka.streams.StreamsConfig.ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG;
import static org.apache.kafka.streams.StreamsConfig.PROCESSOR_WRAPPER_CLASS_CONFIG;
+
Review Comment:
```suggestion
```
##########
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java:
##########
@@ -384,7 +384,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materializedInternal =
new MaterializedInternal<>(
- Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()),
+ Materialized.<K, V, KeyValueStore<Bytes, byte[]>>with(consumedInternal.keySerde(), consumedInternal.valueSerde()).withLoggingDisabled(),
Review Comment:
Ah, yes. Makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1965533413
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
Sounds good, trying to shorten it a bit (just an idea):
```
Explicit naming for internal resources is currently disabled. If you want to enforce user-defined names for all internal resources, set "ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG" to true. Note: Changing internal resource names may require a full streams application reset for an already deployed application. Consult the documentation on naming operators for more details.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1965535691
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
sounds goods to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #18233:
URL: https://github.com/apache/kafka/pull/18233
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1956163352
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,11 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal " +
+ " topics (e.g., changelog and repartition topics) and their associated state stores." +
+ " When enabled, the application will refuse to start if any internal resource has an auto-generated name.";
/** {@code log.summary.interval.ms} */
Review Comment:
done
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,11 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal " +
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1956770352
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
I don't necessarily have a problem with the warning log in theory, but I think we need to be careful to specify that you shouldn't add names if they have been missing from the start. This warning is really only useful if people see it before deploying their app the first time.
I guess it might still help them remember to add names the next time they write an app, but we should definitely note this in the log. But the current wording could potentially suggest they should stop and add names to everything which will mess up an app that already has data in the changelogs/repartitions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2548834041
There are some missing unit tests. I will add them, especially once I am confident that the implementation is on the right track.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2561582697
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937326984
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2333,4 +2341,33 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addUnprovidedInternalTopics(final InternalResourcesNaming internalResourcesNaming) {
Review Comment:
changed to addImplicitInternalNames
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1965507035
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
@lucasbru @ableegoldman
What do you think about that:
Explicit naming for internal resources is currently disabled. If you want to enforce user-defined names for all internal resources, set `" ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG"` to `true`.
Note: If your application has already been deployed with auto-generated names, changing internal resource names may impact compatibility.
If you need to apply this configuration to an existing application, consider resetting the Streams application using the Kafka Streams reset tool."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937323264
##########
streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java:
##########
@@ -592,6 +592,10 @@ public class StreamsConfig extends AbstractConfig {
public static final String ENABLE_METRICS_PUSH_DOC = "Whether to enable pushing of internal client metrics for (main, restore, and global) consumers, producers, and admin clients." +
" The cluster must have a client metrics subscription which corresponds to a client.";
+ /** {@code ensure.explicit.internal.resource.naming} */
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG = "ensure.explicit.internal.resource.naming";
+ public static final String ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_DOC = "Whether to enforce explicit naming for all internal resources of the topology, including internal topics (e.g., changelog and repartition topics) and their associated state stores." +
+ " When enabled, the application will fail to start if any internal resource has an auto-generated name.";
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937350055
##########
streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java:
##########
@@ -384,7 +384,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
final MaterializedInternal<K, V, KeyValueStore<Bytes, byte[]>> materializedInternal =
new MaterializedInternal<>(
- Materialized.with(consumedInternal.keySerde(), consumedInternal.valueSerde()),
+ Materialized.<K, V, KeyValueStore<Bytes, byte[]>>with(consumedInternal.keySerde(), consumedInternal.valueSerde()).withLoggingDisabled(),
Review Comment:
For a GlobalKTable logging is always disable, I have to specify it here otherwise a changelog topic will be added as a not named internal resource:
https://github.com/sebastienviale/kafka/blob/a8790a02ff1f1de7d387b1cdaf9b93b349e6403e/streams/src/main/java/org/apache/kafka/streams/kstream/internals/MaterializedInternal.java#L54-L64
what do you think ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1937429178
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/CogroupedStreamAggregateBuilder.java:
##########
@@ -204,7 +204,8 @@ <KR> KTable<KR, VOut> build(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? sup
}
private void processRepartitions(final Map<KGroupedStreamImpl<K, ?>, Aggregator<? super K, ? super Object, VOut>> groupPatterns,
Review Comment:
last argument was not used anymore after rebase
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #18233:
URL: https://github.com/apache/kafka/pull/18233#issuecomment-2674663120
@sebastienviale Your changes to the docs look good to me. I wonder if it would also make sense to add a little paragraph to docs/streams/developer-guide/dsl-topology-naming.html where it makes sense? Just to point the reader to the new config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1956164410
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java:
##########
@@ -1051,76 +1063,15 @@ public <VTable, VOut> KStream<K, VOut> leftJoin(final KTable<K, VTable> table,
final KStreamImpl<K, V> thisStreamRepartitioned = repartitionForJoin(
name != null ? name : this.name,
joinedInternal.keySerde(),
- joinedInternal.leftValueSerde()
+ joinedInternal.leftValueSerde(),
+ name != null
);
return thisStreamRepartitioned.doStreamTableJoin(table, joiner, joinedInternal, true);
} else {
return doStreamTableJoin(table, joiner, joinedInternal, true);
}
}
- @SuppressWarnings({"unchecked", "resource"})
Review Comment:
I must have made a mistake when I rebased
done
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImplJoin.java:
##########
@@ -168,6 +169,15 @@ public <K, V1, V2, VOut> KStream<K, VOut> join(final KStream<K, V1> lhs,
);
}
+ if (userProvidedBaseStoreName == null) {
+ addInternalResourceName(thisWindowStore);
+ addInternalResourceName(otherWindowStore);
+ if (outerJoinWindowStore.isPresent()) {
+ addInternalResourceName(outerJoinWindowStore.get());
+ }
+
+ }
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1956164902
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -64,12 +66,15 @@
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import static org.apache.kafka.streams.StreamsConfig.ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG;
import static org.apache.kafka.streams.StreamsConfig.PROCESSOR_WRAPPER_CLASS_CONFIG;
+
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18023: Enforcing Explicit Naming for Kafka Streams Internal Topics [kafka]
Posted by "sebastienviale (via GitHub)" <gi...@apache.org>.
sebastienviale commented on code in PR #18233:
URL: https://github.com/apache/kafka/pull/18233#discussion_r1956165435
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java:
##########
@@ -2289,4 +2298,45 @@ public <KIn, VIn, KOut, VOut> WrappedProcessorSupplier<KIn, VIn, KOut, VOut> wra
processorWrapper.wrapProcessorSupplier(name, processorSupplier)
);
}
+
+ public void addImplicitInternalNames(final InternalResourcesNaming internalResourcesNaming) {
+ implicitInternalNames.add(internalResourcesNaming);
+ }
+
+ public void checkUnprovidedNames() {
+ if (!implicitInternalNames.isEmpty()) {
+ final StringBuilder result = new StringBuilder();
+ final List<String> changelogTopics = new ArrayList<>();
+ final List<String> stateStores = new ArrayList<>();
+ final List<String> repartitionTopics = new ArrayList<>();
+ for (final InternalResourcesNaming internalResourcesNaming : implicitInternalNames) {
+ if (!Utils.isBlank(internalResourcesNaming.changelogTopic())) {
+ changelogTopics.add(internalResourcesNaming.changelogTopic());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.stateStore())) {
+ stateStores.add(internalResourcesNaming.stateStore());
+ }
+ if (!Utils.isBlank(internalResourcesNaming.repartitionTopic())) {
+ repartitionTopics.add(internalResourcesNaming.repartitionTopic());
+ }
+ }
+ if (!changelogTopics.isEmpty()) {
+ result.append(String.format("Following changelog topic(s) has not been named: %s%n", String.join(", ", changelogTopics)));
+ }
+ if (!stateStores.isEmpty()) {
+ result.append(String.format("Following state store(s) has not been named: %s%n", String.join(", ", stateStores)));
+ }
+ if (!repartitionTopics.isEmpty()) {
+ result.append(String.format("Following repartition topic(s) has not been named: %s%n", String.join(", ", repartitionTopics)));
+ }
+ if (ensureExplicitInternalResourceNaming) {
+ throw new TopologyException(result.toString());
+ } else {
+ log.warn("Enforce explicit naming for all internal resources is set to false. If you want" +
Review Comment:
let's wait then !
--
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