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 2022/01/24 09:25:14 UTC
[GitHub] [kafka] showuon opened a new pull request #11705: [WIP] KAFKA-9847: add config to set default store type
showuon opened a new pull request #11705:
URL: https://github.com/apache/kafka/pull/11705
WIP
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r811658823
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedCogroupedKStreamImpl.java
##########
@@ -113,14 +114,23 @@
+ "]");
}
- supplier = Stores.persistentTimestampedWindowStore(
- materialized.storeName(),
- Duration.ofMillis(retentionPeriod),
- Duration.ofMillis(windows.timeDifferenceMs()),
- false
- );
-
+ if (materialized.storeType().equals(Materialized.StoreType.IN_MEMORY)) {
+ supplier = Stores.inMemoryWindowStore(
Review comment:
Yes, from the [java doc](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/Stores.java#L264-L267) in the `inMemoryWindowStore`, it said:
```
* Create an in-memory {@link
* This store supplier can be passed into a {@link #windowStoreBuilder(WindowBytesStoreSupplier, Serde, Serde)} or
* {@link #timestampedWindowStoreBuilder(WindowBytesStoreSupplier, Serde, Serde)}.
```
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r811681174
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
@ableegoldman , Sorry, I don't understand what you mean here. Are you referring this [newNamedTopologyBuilder](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java#L116-L129) method? Or should I open another JIRA ticket for 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] guozhangwang commented on pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1067276610
Re-triggered jenkins.
--
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 pull request #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1047556093
@guozhangwang , I've updated the PR in this commit: https://github.com/apache/kafka/pull/11705/commits/d6a033d0c063f4b27c1aa6d46675e140e80f75e9. 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
[GitHub] [kafka] showuon commented on pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1069884347
Thanks for all your review!
--
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 merged pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon merged pull request #11705:
URL: https://github.com/apache/kafka/pull/11705
--
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] mjsax commented on a change in pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
mjsax commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827380865
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -153,6 +166,17 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
} else {
deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+
+ // Override the default store type config no matter it's a named topology or not since it should apply to all topologies
+ storeType = getString(DEFAULT_DSL_STORE_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DSL_STORE_CONFIG, storeType);
Review comment:
Why do we log this unconditionally? Should we not have a overwrite for `isTopologyOverride`?
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/MaterializedInternalTest.java
##########
@@ -73,4 +79,24 @@ public void shouldUseStoreNameOfSupplierWhenProvided() {
new MaterializedInternal<>(Materialized.as(supplier), nameProvider, prefix);
assertThat(materialized.storeName(), equalTo(storeName));
}
+
+ @Test
+ public void shouldUseStoreTypeWhenProvidedViaTopologyConfig() {
+ final Properties topologyOverrides = new Properties();
+ topologyOverrides.put(StreamsConfig.DEFAULT_DSL_STORE_CONFIG, StreamsConfig.IN_MEMORY);
+ final StreamsConfig config = new StreamsConfig(StreamsTestUtils.getStreamsConfig());
+
+ final InternalTopologyBuilder topologyBuilder = new InternalTopologyBuilder(
+ new TopologyConfig(
+ "my-topology",
Review comment:
nit: indention
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/Materialized.java
##########
@@ -259,4 +285,16 @@ protected Materialized(final Materialized<K, V, S> materialized) {
this.retention = retention;
return this;
}
+
+ /**
+ * Set the type of the materialized {@link StateStore}.
+ *
+ * @param storeType the store type {@link StoreType} to use.
+ * @return itself
+ */
+ public Materialized<K, V, S> withStoreType(final StoreType storeType) throws IllegalArgumentException {
+ Objects.requireNonNull(storeType, "store type can't be null");
Review comment:
Should we set `storeSupplier = null` here? Or throw an exception (or log a WARN)? It seems not to make sense to have both `storeType` and `storeSupplier` being set?
##########
File path: streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
##########
@@ -1164,15 +1172,19 @@ public void kGroupedStreamZeroArgCountShouldPreserveTopologyStructure() {
" <-- KSTREAM-SOURCE-0000000000\n\n",
describe.toString()
);
+
+ assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
Review comment:
Not sure if we should piggy-back this onto an existing test?
If we want to by systematic, it might be better to parametrize the test and let all tests re-run with "default" as well as setting RocksDB and in-memory expliclity (either via config or via parameter call)?
It seems random to set RocksDB for some tests and in-memory for others?
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
What is the purpose of this method? Seems the constructor could just be called directly? Should we remove it?
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImpl.java
##########
@@ -121,11 +121,22 @@
+ " grace=[" + sessionWindows.gracePeriodMs() + "],"
+ " retention=[" + retentionPeriod + "]");
}
- supplier = Stores.persistentSessionStore(
- materialized.storeName(),
- Duration.ofMillis(retentionPeriod)
- );
+
+ switch (materialized.storeType()) {
+ case IN_MEMORY:
+ supplier = Stores.inMemorySessionStore(
+ materialized.storeName(),
+ Duration.ofMillis(retentionPeriod)
+ );
+ break;
+ default:
Review comment:
Can we make this a `case ROCK_DB` and add a
```
default:
throw new IllegalStateException("Unknown store type: " + materialized.storeType());
```
Similar elsewhere.
--
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 #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827578212
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -153,6 +166,17 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
} else {
deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+
+ // Override the default store type config no matter it's a named topology or not since it should apply to all topologies
+ storeType = getString(DEFAULT_DSL_STORE_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DSL_STORE_CONFIG, storeType);
Review comment:
Make sense. 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
[GitHub] [kafka] showuon commented on a change in pull request #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r811648912
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/MaterializedInternal.java
##########
@@ -56,6 +69,10 @@ public String storeName() {
return storeName;
}
+ public StoreType storeType() {
+ return storeType == null ? StoreType.ROCKS_DB : storeType;
Review comment:
You're right! We can set default to rocksDB if `null`, and we don't need this null check here. Updated.
--
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 #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827596487
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -53,36 +59,42 @@
private static final ConfigDef CONFIG;
Review comment:
Actually, I think you already made it very clear in the javadoc [here](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java#L46-L51):
> 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 builders via the {@link org.apache.kafka.streams.StreamsBuilder()} method.
Please let me know if you think we still need to add anything. 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
[GitHub] [kafka] guozhangwang commented on pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1070221651
+1, thanks @showuon for the contribution! And congrats for it finally being merged.
--
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 pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1065619186
@guozhangwang @ableegoldman , sorry for the ping. I think this small feature should be good to get into v3.2.0, which feature freeze is on 3/16, if there is no other comments. WDYT?
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
[GitHub] [kafka] guozhangwang commented on a change in pull request #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r805466422
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedCogroupedKStreamImpl.java
##########
@@ -121,11 +121,20 @@
+ " grace=[" + sessionWindows.gracePeriodMs() + "],"
+ " retention=[" + retentionPeriod + "]");
}
- supplier = Stores.persistentSessionStore(
- materialized.storeName(),
- Duration.ofMillis(retentionPeriod)
- );
+
+ if (materialized.storeType().equals(Materialized.StoreType.IN_MEMORY)) {
Review comment:
Could we use `switch-case` instead of `if-else` in case we add more store types in the future? Ditto elsewhere.
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/MaterializedInternal.java
##########
@@ -56,6 +69,10 @@ public String storeName() {
return storeName;
}
+ public StoreType storeType() {
+ return storeType == null ? StoreType.ROCKS_DB : storeType;
Review comment:
The `storeType` should never be null since it's always initialized at construction time right?
##########
File path: streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
##########
@@ -1262,6 +1336,381 @@ public void timeWindowAnonymousMaterializedCountShouldPreserveTopologyStructure(
" <-- KSTREAM-SOURCE-0000000000\n\n",
describe.toString()
);
+
+ assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
+ }
+
+ @Test
+ public void timeWindowAnonymousStoreTypeMaterializedCountShouldPreserveTopologyStructure() {
+ final StreamsBuilder builder = new StreamsBuilder();
+ builder.stream("input-topic")
+ .groupByKey()
+ .windowedBy(TimeWindows.of(ofMillis(1)))
+ .count(Materialized.as(Materialized.StoreType.IN_MEMORY));
+ final Topology topology = builder.build();
+ final TopologyDescription describe = topology.describe();
+ assertEquals(
+ "Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
+ " --> KSTREAM-AGGREGATE-0000000003\n" +
+ " Processor: KSTREAM-AGGREGATE-0000000003 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000002])\n" +
+ " --> none\n" +
+ " <-- KSTREAM-SOURCE-0000000000\n\n",
+ describe.toString()
+ );
+
+ assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
+ }
+
+ @Test
+ public void timeWindowZeroArgCountWithTopologyConfigShouldPreserveTopologyStructure() {
+ // override the default store into in-memory
+ final StreamsBuilder builder = new StreamsBuilder(overrideDefaultStore(StreamsConfig.IN_MEMORY));
+ builder.stream("input-topic")
+ .groupByKey()
+ .windowedBy(TimeWindows.of(ofMillis(1)))
+ .count();
+ final Topology topology = builder.build();
+ final TopologyDescription describe = topology.describe();
+ assertEquals(
+ "Topology: my-topology:\n" +
+ " Sub-topology: 0\n" +
+ " Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])\n" +
+ " --> KSTREAM-AGGREGATE-0000000002\n" +
+ " Processor: KSTREAM-AGGREGATE-0000000002 (stores: [KSTREAM-AGGREGATE-STATE-STORE-0000000001])\n" +
+ " --> none\n" +
+ " <-- KSTREAM-SOURCE-0000000000\n\n",
+ describe.toString()
+ );
+
+ assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(false));
+ }
+
+ @Test
+ public void slidingWindowZeroArgCountShouldPreserveTopologyStructure() {
Review comment:
Thanks for adding coverage for sliding windows.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/TopologyConfig.java
##########
@@ -153,6 +166,17 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
} else {
deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+
+ // Override the default store type config no matter it's a named topology or not since it should apply to all topologies
+ storeType = getString(DEFAULT_DSL_STORE_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DSL_STORE_CONFIG, storeType);
+ }
+
+ public Materialized.StoreType parseStoreType() {
+ if (storeType.equals(IN_MEMORY)) {
Review comment:
Ditto here for using `switch case`.
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedCogroupedKStreamImpl.java
##########
@@ -113,14 +114,23 @@
+ "]");
}
- supplier = Stores.persistentTimestampedWindowStore(
- materialized.storeName(),
- Duration.ofMillis(retentionPeriod),
- Duration.ofMillis(windows.timeDifferenceMs()),
- false
- );
-
+ if (materialized.storeType().equals(Materialized.StoreType.IN_MEMORY)) {
+ supplier = Stores.inMemoryWindowStore(
Review comment:
Just double checking if `inMemoryWindowStore` maps to timestamped-window store too?
--
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 #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827565592
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -53,36 +59,42 @@
private static final ConfigDef CONFIG;
Review comment:
Maybe make a note in the javadocs that settings values for anything in TopologyConfig will result in that overriding the value set in the StreamsConfig passed in to the KafkaStreams. The current javadocs kind of hint at it but they're more focused on named topologies so I don't think it's really that clear
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r808518197
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
Not specific to this PR since it looks like we already have the TopologyConfig class as a parameter here, but if it's going to be in the public API then we need to move it out of the current package ie the `internals.namedtopology` one. Because it's definitely a public API if it's going to be a parameter of some other public API, and we don't want users to think that everything under the `internals.namedtopology` is also fair game since we haven't done the KIP for it yet
cc @guozhangwang @showuon @wcarlson5
--
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 pull request #11705: [WIP] KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1022206850
@mjsax , I know the KIP is still having discussion about the `Stores` API. But I think the PR is ready for pre-review right now (if you have time :)).
For the `Stores` API change, I'll update it after we reach a conclusion.
Thank you.
--
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] lihaosky commented on a change in pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
lihaosky commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r826228402
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -153,6 +166,17 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
} else {
deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+
+ // Override the default store type config no matter it's a named topology or not since it should apply to all topologies
+ storeType = getString(DEFAULT_DSL_STORE_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DSL_STORE_CONFIG, storeType);
+ }
+
+ public Materialized.StoreType parseStoreType() {
+ if (storeType.equals(IN_MEMORY)) {
Review comment:
nit: curious why `storeType` is not enum but raw string. Typo in `in_memory` could make it `ROCKS_DB` type. e.g. `in_memry`. Maybe it's register in `in(ROCKS_DB, IN_MEMORY),` when defining and checked there
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1051037462
I made another pass and it LGTM. @ableegoldman would you like to make another pass?
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r808708709
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
@ableegoldman , just to confirm, you mean [this constructor](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java#L72-L75), right?
If so, I agree we should remove it after we introduced the public API in `StreamBuilder`.
--
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 #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827562746
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
@mjsax it's used by the named topology feature -- there is a `NamedTopologyBuilder` class that extends `StreamsBuilder` and overrides this method to return a `NamedTopology`.
I know it's a bit awkward, I'll be able to clean it up once I've done the KIP for this feature (which is, finally, in progress)
--
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 edited a comment on pull request #11705: [WIP] KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon edited a comment on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1022206850
@mjsax , I know the KIP is still under discussion about the `Stores` API. But I think the PR is ready for pre-review right now (if you have time :)).
For the `Stores` API change, I'll update it after we reach a conclusion.
Thank you.
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1032215108
Thanks @showuon , making a pass soon.
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r808708709
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
@ableegoldman , just to confirm, you mean [this constructor](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java#L72-L75), right?
If so, I agree we should remove it after we introduced the public API in `StreamBuilder`.
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r811658823
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/SlidingWindowedCogroupedKStreamImpl.java
##########
@@ -113,14 +114,23 @@
+ "]");
}
- supplier = Stores.persistentTimestampedWindowStore(
- materialized.storeName(),
- Duration.ofMillis(retentionPeriod),
- Duration.ofMillis(windows.timeDifferenceMs()),
- false
- );
-
+ if (materialized.storeType().equals(Materialized.StoreType.IN_MEMORY)) {
+ supplier = Stores.inMemoryWindowStore(
Review comment:
Yes, from the [java doc](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/Stores.java#L264-L267) in the `inMemoryWindowStore`, it said:
```
* Create an in-memory {@link WindowBytesStoreSupplier}
* This store supplier can be passed into a {@link #windowStoreBuilder(WindowBytesStoreSupplier, Serde, Serde)} or
* {@link #timestampedWindowStoreBuilder(WindowBytesStoreSupplier, Serde, Serde)}.
```
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r811681174
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
@ableegoldman , Sorry, I don't understand what you mean here. Are you referring this [newNamedTopologyBuilder](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java#L116-L129) method? Or should I open another JIRA ticket for it? 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
[GitHub] [kafka] showuon commented on a change in pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827598127
##########
File path: streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
##########
@@ -1164,15 +1172,19 @@ public void kGroupedStreamZeroArgCountShouldPreserveTopologyStructure() {
" <-- KSTREAM-SOURCE-0000000000\n\n",
describe.toString()
);
+
+ assertThat(topology.internalTopologyBuilder.setApplicationId("test").buildTopology().hasPersistentLocalStore(), is(true));
Review comment:
I agree! But since today is the end of v3.2.0 feature freeze date, I created [KAFKA-13747](https://issues.apache.org/jira/browse/KAFKA-13747) for this test refactoring. Will work on it later. Thank you.
--
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 #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827599220
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/Materialized.java
##########
@@ -259,4 +285,16 @@ protected Materialized(final Materialized<K, V, S> materialized) {
this.retention = retention;
return this;
}
+
+ /**
+ * Set the type of the materialized {@link StateStore}.
+ *
+ * @param storeType the store type {@link StoreType} to use.
+ * @return itself
+ */
+ public Materialized<K, V, S> withStoreType(final StoreType storeType) throws IllegalArgumentException {
+ Objects.requireNonNull(storeType, "store type can't be null");
Review comment:
Good suggestion! Updated and tests added.
--
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 pull request #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1030618945
@mjsax @guozhangwang , this PR is ready for review. Thank you.
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r813252418
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
I think the point is that `TopologyConfig` is currently still in `org.apache.kafka.streams.processor.internals.namedtopology`, which is not considered a public API. So for the public class `StreamsBuilder` to take it as a param, we need to move `TopologyConfig` out of that package in this PR, potentially to `org.apache.kafka.streams` as with `StreamsConfig` and `StreamsBuilder` classes.
I thought this is included as part of KIP-591 to make `TopologyConfig` as a public API (note in [KAFKA-13281](https://issues.apache.org/jira/browse/KAFKA-13281), we did not make it a public API yet), as "Currently, the internal implementation for topology-config is all set. In this KIP, we'll publish the API to allow pass the topology-config into StreamBuilder ." 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] showuon commented on pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#issuecomment-1068698914
@ableegoldman @mjsax , thanks for your comments. I've addressed them in this commit: https://github.com/apache/kafka/pull/11705/commits/7bf2a54b26aa823bc570904ed08d19fcb57597ff . Thank you.
--
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 #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r827564609
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -153,6 +166,17 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
} else {
deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+
+ // Override the default store type config no matter it's a named topology or not since it should apply to all topologies
+ storeType = getString(DEFAULT_DSL_STORE_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DSL_STORE_CONFIG, storeType);
Review comment:
+1 -- also note that this sort of only makes sense when using the named topology feature, as otherwise you don't have multiple topologies and not really any reason to set any of these configs differently in the TopologyConfigs vs StreamsConfig passed in to the KafkaStreams constructor.
That said, it will definitely happen, so I guess we should check for overrides whether the topology is named or not. Maybe you can use the `isTopologyOverride` method, and remove the check for `namedTopology ~= null`?
ie, you can/should use `isTopologyOverride`
--
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] lihaosky commented on a change in pull request #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
lihaosky commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r826228402
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -153,6 +166,17 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
} else {
deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+
+ // Override the default store type config no matter it's a named topology or not since it should apply to all topologies
+ storeType = getString(DEFAULT_DSL_STORE_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DSL_STORE_CONFIG, storeType);
+ }
+
+ public Materialized.StoreType parseStoreType() {
+ if (storeType.equals(IN_MEMORY)) {
Review comment:
nit: curious why `storeType` is not enum but raw string. Typo in `in_memory` could make it `ROCKS_DB` type. e.g. `in_memry`.
--
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 #11705: KAFKA-9847: add config to set default store type (KIP-591)
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r826615748
##########
File path: streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java
##########
@@ -153,6 +166,17 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
} else {
deserializationExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, DeserializationExceptionHandler.class);
}
+
+ // Override the default store type config no matter it's a named topology or not since it should apply to all topologies
+ storeType = getString(DEFAULT_DSL_STORE_CONFIG);
+ log.info("Topology {} is overriding {} to {}", topologyName, DEFAULT_DSL_STORE_CONFIG, storeType);
+ }
+
+ public Materialized.StoreType parseStoreType() {
+ if (storeType.equals(IN_MEMORY)) {
Review comment:
@lihaosky , thanks for your comment. `storeType` is retrieved from stream config, where we can't accept enum. Currently, we only accept `rocksDB` or `in_memory`. If there's a typo, the config will throw exception, like we only accept `delete` and `compact` in `log.cleanup.policy`, and will throw exception if the provided value is other values.
--
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 #11705: KAFKA-9847: add config to set default store type
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11705:
URL: https://github.com/apache/kafka/pull/11705#discussion_r814753767
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -81,14 +81,19 @@ public StreamsBuilder() {
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
- protected StreamsBuilder(final TopologyConfig topologyConfigs) {
+ /**
+ * Create a {@code StreamsBuilder} instance.
+ *
+ * @param topologyConfigs the streams configs that apply at the topology level. Please refer to {@link TopologyConfig} for more detail
+ */
+ public StreamsBuilder(final TopologyConfig topologyConfigs) {
topology = getNewTopology(topologyConfigs);
internalTopologyBuilder = topology.internalTopologyBuilder;
internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
}
protected Topology getNewTopology(final TopologyConfig topologyConfigs) {
Review comment:
Yes, agree with you. This should be part of KIP-591 to make the `TopologyConfig` as public API. Moved out of `internals.namedtopology` package now. 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