You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ableegoldman (via GitHub)" <gi...@apache.org> on 2024/12/05 02:55:14 UTC
[PR] KAFKA-18026: KIP-1112, migrate stream-table joins to use ProcesserSupplier#stores [kafka]
ableegoldman opened a new pull request, #18047:
URL: https://github.com/apache/kafka/pull/18047
Covers wrapping of processors and state stores for KStream-KTable joins
--
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-18026: KIP-1112, migrate stream-table joins to use ProcesserSupplier#stores [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #18047:
URL: https://github.com/apache/kafka/pull/18047#discussion_r1870578036
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java:
##########
@@ -32,18 +36,28 @@ class KStreamKTableJoin<K, V1, V2, VOut> implements ProcessorSupplier<K, V1, K,
private final boolean leftJoin;
private final Optional<Duration> gracePeriod;
private final Optional<String> storeName;
-
+ private final Optional<StoreBuilder<?>> storeBuilder;
KStreamKTableJoin(final KTableValueGetterSupplier<K, V2> valueGetterSupplier,
final ValueJoinerWithKey<? super K, ? super V1, ? super V2, VOut> joiner,
final boolean leftJoin,
final Optional<Duration> gracePeriod,
- final Optional<String> storeName) {
+ final Optional<StoreBuilder<?>> storeBuilder) {
this.valueGetterSupplier = valueGetterSupplier;
this.joiner = joiner;
this.leftJoin = leftJoin;
this.gracePeriod = gracePeriod;
- this.storeName = storeName;
+ this.storeName = storeBuilder.map(StoreBuilder::name);
+ this.storeBuilder = storeBuilder;
+ }
+
+ @Override
+ public Set<StoreBuilder<?>> stores() {
Review Comment:
sorry, can you elaborate on what you mean by "this"? Like wrapping it in the Collections.singleton to convert it to a Set<StoreBuilder>?
--
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-18026: KIP-1112, migrate stream-table joins to use ProcesserSupplier#stores [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman merged PR #18047:
URL: https://github.com/apache/kafka/pull/18047
--
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-18026: KIP-1112, migrate stream-table joins to use ProcesserSupplier#stores [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #18047:
URL: https://github.com/apache/kafka/pull/18047#discussion_r1870585681
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java:
##########
@@ -32,18 +36,28 @@ class KStreamKTableJoin<K, V1, V2, VOut> implements ProcessorSupplier<K, V1, K,
private final boolean leftJoin;
private final Optional<Duration> gracePeriod;
private final Optional<String> storeName;
-
+ private final Optional<StoreBuilder<?>> storeBuilder;
KStreamKTableJoin(final KTableValueGetterSupplier<K, V2> valueGetterSupplier,
final ValueJoinerWithKey<? super K, ? super V1, ? super V2, VOut> joiner,
final boolean leftJoin,
final Optional<Duration> gracePeriod,
- final Optional<String> storeName) {
+ final Optional<StoreBuilder<?>> storeBuilder) {
this.valueGetterSupplier = valueGetterSupplier;
this.joiner = joiner;
this.leftJoin = leftJoin;
this.gracePeriod = gracePeriod;
- this.storeName = storeName;
+ this.storeName = storeBuilder.map(StoreBuilder::name);
+ this.storeBuilder = storeBuilder;
+ }
+
+ @Override
+ public Set<StoreBuilder<?>> stores() {
Review Comment:
gotcha, yeah we can do that
--
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-18026: KIP-1112, migrate stream-table joins to use ProcesserSupplier#stores [kafka]
Posted by "agavra (via GitHub)" <gi...@apache.org>.
agavra commented on code in PR #18047:
URL: https://github.com/apache/kafka/pull/18047#discussion_r1870567816
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java:
##########
@@ -32,18 +36,28 @@ class KStreamKTableJoin<K, V1, V2, VOut> implements ProcessorSupplier<K, V1, K,
private final boolean leftJoin;
private final Optional<Duration> gracePeriod;
private final Optional<String> storeName;
-
+ private final Optional<StoreBuilder<?>> storeBuilder;
KStreamKTableJoin(final KTableValueGetterSupplier<K, V2> valueGetterSupplier,
final ValueJoinerWithKey<? super K, ? super V1, ? super V2, VOut> joiner,
final boolean leftJoin,
final Optional<Duration> gracePeriod,
- final Optional<String> storeName) {
+ final Optional<StoreBuilder<?>> storeBuilder) {
this.valueGetterSupplier = valueGetterSupplier;
this.joiner = joiner;
this.leftJoin = leftJoin;
this.gracePeriod = gracePeriod;
- this.storeName = storeName;
+ this.storeName = storeBuilder.map(StoreBuilder::name);
+ this.storeBuilder = storeBuilder;
+ }
+
+ @Override
+ public Set<StoreBuilder<?>> stores() {
Review Comment:
I think `stores()` is only ever called once, so it's probably fine doing it here, but just wondering why you chose to do this here instead of in the constructor given it always returns the same thing?
--
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-18026: KIP-1112, migrate stream-table joins to use ProcesserSupplier#stores [kafka]
Posted by "agavra (via GitHub)" <gi...@apache.org>.
agavra commented on code in PR #18047:
URL: https://github.com/apache/kafka/pull/18047#discussion_r1870579036
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoin.java:
##########
@@ -32,18 +36,28 @@ class KStreamKTableJoin<K, V1, V2, VOut> implements ProcessorSupplier<K, V1, K,
private final boolean leftJoin;
private final Optional<Duration> gracePeriod;
private final Optional<String> storeName;
-
+ private final Optional<StoreBuilder<?>> storeBuilder;
KStreamKTableJoin(final KTableValueGetterSupplier<K, V2> valueGetterSupplier,
final ValueJoinerWithKey<? super K, ? super V1, ? super V2, VOut> joiner,
final boolean leftJoin,
final Optional<Duration> gracePeriod,
- final Optional<String> storeName) {
+ final Optional<StoreBuilder<?>> storeBuilder) {
this.valueGetterSupplier = valueGetterSupplier;
this.joiner = joiner;
this.leftJoin = leftJoin;
this.gracePeriod = gracePeriod;
- this.storeName = storeName;
+ this.storeName = storeBuilder.map(StoreBuilder::name);
+ this.storeBuilder = storeBuilder;
+ }
+
+ @Override
+ public Set<StoreBuilder<?>> stores() {
Review Comment:
I mean instead of storing the `Optional<StoreBuilder>` we can store `Set<StoreBuilder>` and just run this same code here you have in `stores` but do it in the constructor
--
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