You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/08/20 12:53:47 UTC
[GitHub] [kafka] showuon opened a new pull request #11242: [WIP] MINOR: POC for kIP-591
showuon opened a new pull request #11242:
URL: https://github.com/apache/kafka/pull/11242
### 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 pull request #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-902703961
@vvcephei @mjsax , could you provide suggestion for this POC when available? 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 #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r692951450
##########
File path: streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
##########
@@ -1168,7 +1169,32 @@ public void kGroupedStreamZeroArgCountShouldPreserveTopologyStructure() {
@Test
public void kGroupedStreamNamedMaterializedCountShouldPreserveTopologyStructure() {
- final StreamsBuilder builder = new StreamsBuilder();
+ Properties streamsProp = new Properties();
+ streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-config-test");
+ streamsProp.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+ final StreamsBuilder builder = new StreamsBuilder(streamsProp);
+ builder.stream("input-topic")
+ .groupByKey()
+ .count(Materialized.as("count-store"));
+ final TopologyDescription describe = builder.build().describe();
Review comment:
An example we put the `stream props` into `StreamsBuilder` constructor and build with default store implementation setting for `KeyValueBytesStoreSupplier`. And the same for `WindowBytesStoreSupplier` and `SessionBytesStoreSupplier` below
--
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 closed pull request #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon closed pull request #11242:
URL: https://github.com/apache/kafka/pull/11242
--
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 #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r692947806
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/Materialized.java
##########
@@ -103,6 +110,21 @@ protected Materialized(final Materialized<K, V, S> materialized) {
return new Materialized<>(storeName);
}
+ /**
+ * Materialize a {@link StateStore} with the store implementation.
+ *
+ * @param storeImplementation store implementation used to materialize the store
+ * alphanumerics, '.', '_' and '-'.
+ * @param <K> key type of the store
+ * @param <V> value type of the store
+ * @param <S> type of the {@link StateStore}
+ * @return a new {@link Materialized} instance with the given storeName
+ */
+ public static <K, V, S extends StateStore> Materialized<K, V, S> as(final StoreImplementation storeImplementation) {
Review comment:
new overloading method for `Materialized#as`
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-1020885807
PR for the implementation: 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] showuon commented on a change in pull request #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r692952035
##########
File path: streams/src/test/java/org/apache/kafka/streams/TopologyTest.java
##########
@@ -1185,6 +1211,29 @@ public void kGroupedStreamNamedMaterializedCountShouldPreserveTopologyStructure(
);
}
+ @Test
+ public void kGroupedStreamNamedMaterializedCountWithCustomStoreImplementation() {
+ Properties streamsProp = new Properties();
+ streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-config-test");
+ streamsProp.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+ final StreamsBuilder builder = new StreamsBuilder(streamsProp);
+ builder.stream("input-topic")
+ .groupByKey()
+ .count(Materialized.as(new InMemoryStoreImplementation()));
Review comment:
The example to put in the custom store implementation dynamically.
--
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 closed pull request #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon closed pull request #11242:
URL: https://github.com/apache/kafka/pull/11242
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-979793930
@guozhangwang , thanks for your comment. I finally got time to revisit this POC. Yes, I agree that this new state store config can be defined on the newly introduced `TopologyConfig`. I checked it again, and it'll work well with `TopologyConfig`.
> I thought we would use a enum, as the KIP suggested in enum StoreType, instead of a StoreImplementation interface.
Actually, the reason why we don't choose `enum` is because that only works for our built-in state stores, not for **custom state store**. So, with the `StoreImplementation` interface, user can implement their state stores their, and set as default stores if they want.
I'll start to draft the KIP. Thank you very much!
--
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 #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r692950023
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/SessionWindowedKStreamImpl.java
##########
@@ -237,10 +238,21 @@
+ " grace=[" + windows.gracePeriodMs() + "],"
+ " retention=[" + retentionPeriod + "]");
}
- supplier = Stores.persistentSessionStore(
- materialized.storeName(),
- Duration.ofMillis(retentionPeriod)
- );
+
+ // get from default store implementation
+ StoreImplementation storeImplementation = materialized.storeImplementation();
+ if (storeImplementation != null) {
+ supplier = storeImplementation.sessionBytesStoreSupplier(
+ materialized.storeName(),
+ Duration.ofMillis(retentionPeriod));
+ } else {
+ // fall back to use Rocks Db by default if no store implementation provided
Review comment:
Originally, when storeSupplier in materialized is null, we'll use `Rocks DB`. Now, we'll honor the `storeImplementation` setting (if we got props)
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r757778305
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/Materialized.java
##########
@@ -103,6 +110,21 @@ protected Materialized(final Materialized<K, V, S> materialized) {
return new Materialized<>(storeName);
}
+ /**
+ * Materialize a {@link StateStore} with the store implementation.
+ *
+ * @param storeImplementation store implementation used to materialize the store
+ * alphanumerics, '.', '_' and '-'.
+ * @param <K> key type of the store
+ * @param <V> value type of the store
+ * @param <S> type of the {@link StateStore}
+ * @return a new {@link Materialized} instance with the given storeName
+ */
+ public static <K, V, S extends StateStore> Materialized<K, V, S> as(final StoreImplementation storeImplementation) {
Review comment:
The reason I added this method is because user can still override the default config via this method if they want.
--
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 #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r692947020
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
##########
@@ -387,6 +389,12 @@
public static final String DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG = "default.production.exception.handler";
private static final String DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_DOC = "Exception handling class that implements the <code>org.apache.kafka.streams.errors.ProductionExceptionHandler</code> interface.";
+ /** {@code default.store.impl.class} */
+ @SuppressWarnings("WeakerAccess")
+ public static final String DEFAULT_STORE_IMPLEMENTATION_CLASS_CONFIG = "default.store.impl.class";
+ private static final String DEFAULT_STORE_IMPLEMENTATION_CLASS_DOC = "Store supplier implementation class to use. Default is set as RocksDBStoreImplementation. " +
+ "It can be overwritten dynamically during streaming operation.";
Review comment:
(1) create a stream config: default.store.impl.class to store the default store implementation class, and default to RocksDB
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-906431308
@mjsax @vvcephei @guozhangwang , could you have a look at this POC when available? I need your advice before I can update the KIP and continue to the DISCUSS/VOTE process. 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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r696592661
##########
File path: streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
##########
@@ -1066,12 +1068,22 @@ private Topology getStatefulTopology(final String inputTopic,
final String storeName,
final String globalStoreName,
final boolean isPersistentStore) {
+// final StoreBuilder<KeyValueStore<String, Long>> storeBuilder = Stores.keyValueStoreBuilder(
+// isPersistentStore ?
+// Stores.persistentKeyValueStore(storeName)
+// : Stores.inMemoryKeyValueStore(storeName),
+// Serdes.String(),
+// Serdes.Long());
+
+ // an example to use store implementation to create the store builder for processor API
final StoreBuilder<KeyValueStore<String, Long>> storeBuilder = Stores.keyValueStoreBuilder(
isPersistentStore ?
- Stores.persistentKeyValueStore(storeName)
- : Stores.inMemoryKeyValueStore(storeName),
+ new RocksDBStoreImplementation()
+ : new InMemoryStoreImplementation(),
+ storeName,
Serdes.String(),
Serdes.Long());
Review comment:
An example to use `StoreImplementation` to create the store builder for processor API.
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-914789772
Thanks for the POC @showuon ! I made a pass and here are some high-level thoughts:
1) On the PAPI layer, users today can only add stores via Topology#addStateStore in which a `StoreBuilder` is required, and from the public APIs they are most likely going to create one from the `Stores` factory. Note that `KeyValueStoreBuilder` is not a public API and hence cannot be really used by PAPI users. In other words, in PAPI today users would have to specify the store types explicitly via the `Topology#addStateStore` API anyways, and that's also why we only consider this KIP for DSL originally. If we want to benefit the PAPI users, we would probably need to add new functions in the `Stores` factory which would not specify if it is persistent or in-memory, but only the type (kv, window, etc). That would require a larger KIP and maybe we can exclude that for this one.
2) I personally would prefer as less public API changes/additions as possible for KIPs, since it's always easier to add later when we found it's not sufficient, than add first and regret later :) Plus with more public APIs it is always getting clumsier. For that, my original thought would be, from user's perspective, if `Materialized` construct does not have a supplier set, then Streams would use the supplier as configured by the config.
I.e. we should work with all thee cases below:
```
// explicitly set the store
stream.groupByKey().count(Materialized.as(Stores.inMemoryKeyValueStore("some-name"));
// only set the store type
stream.groupByKey().count(Materialized.as(StoreType.IN_MEMORY));
// do not set the store type at all, rely on the config
stream.groupByKey().count(Materialized.as("some-name"));
```
I understand that today the config is not available yet during the DSL's StreamsBuilder phase, but only during the build() phase. What I was thinking is that during the StreamsBuidler's topology-construction phase, e.g. inside `materialize()` function, instead of
```
if (supplier == null) {
final String name = materialized.storeName();
supplier = Stores.persistentTimestampedKeyValueStore(name);
}
```
That means, for the third case above, we would need some "placeholder" StoreSupplier as it is "to be decided", and then during `buildAndOptimizeTopology` we would fill-in the placeholders based on the configs. For the other two cases, we would derive the supplier immediately. With that we would not need the `StoreImplementation` interface.
In general I'm leaning towards we, as the developers of KS, eat more complexity bullets so that we can leave the public interface as succinct as possible to the users. As for passing in the config at the `StreamsBuilder` constructor instead of the build phase, I think that would work too, and would be much simpler, but again it would change the public API pattern as well which makes the `build(properties)` function deprecated, and the other deprecated `build()` back to normal? Maybe cc @ableegoldman as well since she's working on another project which would benefit if we pass in the config at the constructor phase).
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-929242849
@mjsax @vvcephei , please take a look when available and let us know your thoughts. 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 #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r692948732
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/MaterializedInternal.java
##########
@@ -29,14 +31,26 @@
private final boolean queryable;
public MaterializedInternal(final Materialized<K, V, S> materialized) {
- this(materialized, null, null);
+ this(materialized, null, null, null);
}
public MaterializedInternal(final Materialized<K, V, S> materialized,
final InternalNameProvider nameProvider,
final String generatedStorePrefix) {
+ this(materialized, nameProvider, generatedStorePrefix, null);
+ }
+
+ public MaterializedInternal(final Materialized<K, V, S> materialized,
+ final InternalNameProvider nameProvider,
+ final String generatedStorePrefix,
+ final StreamsConfig config) {
super(materialized);
+ if (config != null && this.storeSupplier == null && this.storeImplementation != null) {
+ // get default store implementation
+ this.storeImplementation = config.defaultStoreImplementation();
+ }
Review comment:
new constructor for `MaterializedInternal`, to accept one more paremeter: `StreamConfig`, so that we can store the `storeImplementation` setting for later initialization.
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-990344744
> Actually, the reason why we don't choose `enum` is because that only works for our built-in state stores, not for **custom state store**. So, with the `StoreImplementation` interface, user can implement their state stores their, and set as default stores if they want.
I understand that, but I'm still having some reservations about this approach. Let me share it on the DISCUSS thread so that we can engage larger groups.
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-929242849
@mjsax @vvcephei , please take a look when available and let us know your thoughts. 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 #11242: [WIP] MINOR: POC for KIP-591
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r692947480
##########
File path: streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java
##########
@@ -74,10 +74,18 @@
protected final InternalStreamsBuilder internalStreamsBuilder;
+ protected final StreamsConfig config;
+
public StreamsBuilder() {
+ this(null);
+ }
+
+ // new constructor
+ public StreamsBuilder(Properties props) {
topology = getNewTopology();
internalTopologyBuilder = topology.internalTopologyBuilder;
- internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder);
+ config = props == null ? null : new StreamsConfig(props);
+ internalStreamsBuilder = new InternalStreamsBuilder(internalTopologyBuilder, config);
Review comment:
(4) add a new constructor for StreamsBuilder, to accept a properties instance.
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#discussion_r742533219
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java
##########
@@ -224,12 +225,25 @@
+ " retention=[" + retentionPeriod + "]");
}
- supplier = Stores.persistentTimestampedWindowStore(
+ // get from default store implementation
+ final StoreImplementation storeImplementation = materialized.storeImplementation();
Review comment:
This is a meta comment: I feel the introduction of the `StoreImplementation` is a bit overkill here. Could the same be achieved with just an `enum` type say "StoreImplTypes" kept inside the Materialize object, and inside `Stores` the added creator would be:
```
Stores#keyValueBytesStoreSupplier(storeImplTypes, storeName, ...)
Stores#windowBytesStoreSupplier(storeImplTypes, storeName, ...)
Stores#sessionBytesStoreSupplier(storeImplTypes, storeName, ...)
```
Then inside the impl classes like here we would could directly do:
```
if ((supplier = materialized.storeSupplier) == null) {
supplier = Stores.windowBytesStoreSupplier(materialized.storeImplType())
}
```
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/Materialized.java
##########
@@ -103,6 +110,21 @@ protected Materialized(final Materialized<K, V, S> materialized) {
return new Materialized<>(storeName);
}
+ /**
+ * Materialize a {@link StateStore} with the store implementation.
+ *
+ * @param storeImplementation store implementation used to materialize the store
+ * alphanumerics, '.', '_' and '-'.
+ * @param <K> key type of the store
+ * @param <V> value type of the store
+ * @param <S> type of the {@link StateStore}
+ * @return a new {@link Materialized} instance with the given storeName
+ */
+ public static <K, V, S extends StateStore> Materialized<K, V, S> as(final StoreImplementation storeImplementation) {
Review comment:
In practice I thought no users code would call this function since they are all set via the config? I have not seen any non-testing code that calls this.
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java
##########
@@ -224,12 +225,25 @@
+ " retention=[" + retentionPeriod + "]");
}
- supplier = Stores.persistentTimestampedWindowStore(
+ // get from default store implementation
+ final StoreImplementation storeImplementation = materialized.storeImplementation();
+ if (storeImplementation != null) {
+ supplier = storeImplementation.windowBytesStoreSupplier(
+ materialized.storeName(),
+ Duration.ofMillis(retentionPeriod),
+ Duration.ofMillis(windows.size()),
+ false);
+ } else {
Review comment:
Also since we always default to the `persistent` type at the MaterializedInternal constructor if the config is not set, we should never need to hit this fallback logic right? If I read it wrong, could we make sure that at Materialized constructor if the config value is not set (should not be the case since we define its default value indeed?), we then hard-code a persistent type.
--
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 #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-945709251
@mjsax @vvcephei , please take a look when available and let us know your thoughts. 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 pull request #11242: [WIP] MINOR: POC for KIP-591: Add config to set default store impl class
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11242:
URL: https://github.com/apache/kafka/pull/11242#issuecomment-1020885807
PR for the implementation: 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