KAFKA-15774: introduce internal StoreFactory #14659

agavra · 2023-10-27T23:44:20Z

Overview

This PR sets up the necessary prerequisites to respect configurations such as dsl.default.store.type and the dsl.store.suppliers.class (introduced in #14648) without requiring them to be passed in to StreamBuilder#new(TopologyConfig) (passing them only into new KafkaStreams(...).

It essentially makes StoreBuilder an external-only API and internally it uses the StoreFactory equivalent. It replaces KeyValueStoreMaterializer with an implementation of StoreFactory that creates the store builder only after configure() is called (in a Future PR we will create the missing equivalents for all of the places where the same thing happens for window stores, such as in all the *WindowKStreamImpl classes)

Testing

There is no change in functionality for this PR

Review Guide

Start with looking at StoreFactory and read the JavaDocs, this is an interface representing what used to be InternalTopologyBuilder.StateStoreFactory
Look at how StoreBuilderWrapper is what used to be the implementation of InternalTopologyBuilder.StateStoreFactory
Note that InternalTopologyBuilder#addStateStore now takes in a StoreFactory instead of a StoreBuilder from everywhere that uses the DSL.
The rest is piping that change around and wrapping StoreBuilder with StoreBuilderWrapper. In a future PR all StoreBuilderWrappers in the DSL will be replaced with a new one that respects the configurations passed in to new KafkaStreams

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

ableegoldman · 2023-10-31T07:00:06Z

streams/src/main/java/org/apache/kafka/streams/StreamsBuilder.java

@@ -520,7 +521,7 @@ public synchronized <K, V> GlobalKTable<K, V> globalTable(final String topic,
     */
    public synchronized StreamsBuilder addStateStore(final StoreBuilder<?> builder) {
        Objects.requireNonNull(builder, "builder can't be null");
-        internalStreamsBuilder.addStateStore(builder);
+        internalStreamsBuilder.addStateStore(new StoreBuilderWrapper<>(builder));


Right off the bat I don't love the name StoreBuilderWrapper...but I think we both know this was going to be a practice in painful out of control java naming lol

I originally tried DelegatingStoreFactory but settled on this. Open to suggestions :)

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreBuilderWrapper.java

ableegoldman · 2023-10-31T07:21:28Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

            }
        }
        nodeGroups = null;
    }

-    public final <KIn, VIn> void addGlobalStore(final StoreBuilder<?> storeBuilder,
+    public final <KIn, VIn> void addGlobalStore(final StoreFactory<?> storeBuilder,


nit: change the parameter name to storeFactory

yup, there were lots of places I need to make this change but I didn't want to touch 2 million lines of code until the general strategy was looked at 😆

very understandable -- just wanted to leave a comment in case you missed it/to help identify some of those two million lines 😭

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreBuilderWrapper.java

ableegoldman · 2023-10-31T07:25:38Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreBuilderWrapper.java

+    }
+
+    @Override
+    public boolean isCompatibleWith(final StoreFactory<?> storeFactory) {


I know you didn't add this code (even if the method itself is new) but some javadocs or a more descriptive name might be useful here. Hard to understand whether this is/will be working correctly after the changes when it's unclear what its original semantics/purpose were

I guess this is related to a fundamental source of confusion I have right now, which is: where does the StoreBuilderWrapper fit into the final plan? Will anything extend it, or will uninitialized/unconfigured StoreFactory implementations somehow be compiled into this class when they finally set up a real StoreBuilder? Like is this class a "storebuilder-wrapper" in the sense that it always wraps an actual configured/initialized StoreBuilder object, or will the builder field of this class be "delayed" (in which case how do we evaluate if it's compatible, hence how it all relates to this specific question/method)

I'll add a comment. The plan is to add a separate DelayedStoreFactory which delays the creation of the StoreBuilder until configure is called. the isCompatibleWith method will check to make sure the inner components would generate the same store builder (e.g. the wrapped StoreSuppliers are identical)

I'll just make that change and add it to this PR to show how it works end-to-end, but that will be a functionality change.

EDIT: actually I can add it w/o changing functionality if I just ignore configure()

ableegoldman

Just the one small comment for now...looks good on the whole. I assume a similar approach will be taken for window/session stores, though we'll have to implement the Materializer class for each of these since they don't already exist

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java

ableegoldman · 2023-11-03T00:13:44Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreFactory.java

+ *     <li>{@link org.apache.kafka.streams.state.StoreSupplier} is used by the
+ *     DSL to provide preconfigured state stores as well as type-safe stores
+ *     (e.g. {@link org.apache.kafka.streams.state.KeyValueBytesStoreSupplier}.
+ *     Before passing the {@code StoreSupplier} into the </li>


incomplete sentence?

🙄 🤭 😬

ableegoldman · 2023-11-03T00:31:27Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreBuilderWrapper.java

+import org.apache.kafka.streams.state.internals.VersionedKeyValueStoreBuilder;
+import org.apache.kafka.streams.state.internals.WindowStoreBuilder;
+
+public class StoreBuilderWrapper implements StoreFactory {


Ok one more thing: the javadocs explaining the StoreFactory are great, but I think this needs javaodcs as well. Is it just the default implementation of the StoreFactory class? Is it just here to cover the window/session store cases because they don't yet have a Materializer? etc

It's here mostly to cover use cases where the user passes in StoreBuilder directly (via PAPI or DSL#processes funs) - so it's here to stay :)

Aha -- thanks for clarifying. So IIUC it is like a "wrapper" for already-instantiated StoreBuilders? The current name actually makes total sense once I understood what the class does. Of course ideally the reverse is true...that you can understand what the class does based on the name. But maybe it's obvious and I just lack reading comprehension skills 🤷‍♀️

I can't think of a better name, but I feel better now that there's a description for this. Thanks!

ableegoldman · 2023-11-03T23:09:34Z

streams/src/main/java/org/apache/kafka/streams/kstream/Materialized.java

+                    return StoreType.IN_MEMORY;
+                case StreamsConfig.ROCKS_DB:
+                default:
+                    // for backwards compatibility, we ignore invalid store


Honestly I would just make this change now, and throw an exception if it's invalid. If you really want to keep the behavior the same for whatever reason, that's fine too, but we should at least log a warning

agavra · 2023-11-06T16:49:11Z

@ableegoldman none of the failing tests seem remotely related and a quick spotcheck shows them passing locally:

ableegoldman

LGTM

mjsax · 2023-11-08T07:16:04Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java

@@ -16,30 +16,9 @@
 */
 package org.apache.kafka.streams.processor.internals;

-import org.apache.kafka.clients.consumer.OffsetResetStrategy;


Please avoid auto-reformatting.

This is a follow up from #14659 that ports the windowed classes to use the StoreFactory abstraction as well. There's a side benefit of not duplicating the materialization code twice for each StreamImpl/CogroupedStreamImpl class as well. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Matthias Sax <mjsax@apache.org>

This PR sets up the necessary prerequisites to respect configurations such as dsl.default.store.type and the dsl.store.suppliers.class (introduced in apache#14648) without requiring them to be passed in to StreamBuilder#new(TopologyConfig) (passing them only into new KafkaStreams(...). It essentially makes StoreBuilder an external-only API and internally it uses the StoreFactory equivalent. It replaces KeyValueStoreMaterializer with an implementation of StoreFactory that creates the store builder only after configure() is called (in a Future PR we will create the missing equivalents for all of the places where the same thing happens for window stores, such as in all the *WindowKStreamImpl classes) Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>

This is a follow up from apache#14659 that ports the windowed classes to use the StoreFactory abstraction as well. There's a side benefit of not duplicating the materialization code twice for each StreamImpl/CogroupedStreamImpl class as well. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Matthias Sax <mjsax@apache.org>

This PR sets up the necessary prerequisites to respect configurations such as dsl.default.store.type and the dsl.store.suppliers.class (introduced in apache#14648) without requiring them to be passed in to StreamBuilder#new(TopologyConfig) (passing them only into new KafkaStreams(...). It essentially makes StoreBuilder an external-only API and internally it uses the StoreFactory equivalent. It replaces KeyValueStoreMaterializer with an implementation of StoreFactory that creates the store builder only after configure() is called (in a Future PR we will create the missing equivalents for all of the places where the same thing happens for window stores, such as in all the *WindowKStreamImpl classes) Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>

This PR sets up the necessary prerequisites to respect configurations such as dsl.default.store.type and the dsl.store.suppliers.class (introduced in apache#14648) without requiring them to be passed in to StreamBuilder#new(TopologyConfig) (passing them only into new KafkaStreams(...). It essentially makes StoreBuilder an external-only API and internally it uses the StoreFactory equivalent. It replaces KeyValueStoreMaterializer with an implementation of StoreFactory that creates the store builder only after configure() is called (in a Future PR we will create the missing equivalents for all of the places where the same thing happens for window stores, such as in all the *WindowKStreamImpl classes) Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org> Co-authored-by: Almog Gavra <almog.gavra@gmail.com>

This is a follow up from apache#14659 that ports the windowed classes to use the StoreFactory abstraction as well. There's a side benefit of not duplicating the materialization code twice for each StreamImpl/CogroupedStreamImpl class as well. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Matthias Sax <mjsax@apache.org>

…) (#233) This is a follow up from apache#14659 that ports the windowed classes to use the StoreFactory abstraction as well. There's a side benefit of not duplicating the materialization code twice for each StreamImpl/CogroupedStreamImpl class as well. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Matthias Sax <mjsax@apache.org> Co-authored-by: Almog Gavra <almog.gavra@gmail.com>

This PR sets up the necessary prerequisites to respect configurations such as dsl.default.store.type and the dsl.store.suppliers.class (introduced in apache#14648) without requiring them to be passed in to StreamBuilder#new(TopologyConfig) (passing them only into new KafkaStreams(...). It essentially makes StoreBuilder an external-only API and internally it uses the StoreFactory equivalent. It replaces KeyValueStoreMaterializer with an implementation of StoreFactory that creates the store builder only after configure() is called (in a Future PR we will create the missing equivalents for all of the places where the same thing happens for window stores, such as in all the *WindowKStreamImpl classes) Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>

This is a follow up from apache#14659 that ports the windowed classes to use the StoreFactory abstraction as well. There's a side benefit of not duplicating the materialization code twice for each StreamImpl/CogroupedStreamImpl class as well. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Matthias Sax <mjsax@apache.org>

This PR sets up the necessary prerequisites to respect configurations such as dsl.default.store.type and the dsl.store.suppliers.class (introduced in apache#14648) without requiring them to be passed in to StreamBuilder#new(TopologyConfig) (passing them only into new KafkaStreams(...). It essentially makes StoreBuilder an external-only API and internally it uses the StoreFactory equivalent. It replaces KeyValueStoreMaterializer with an implementation of StoreFactory that creates the store builder only after configure() is called (in a Future PR we will create the missing equivalents for all of the places where the same thing happens for window stores, such as in all the *WindowKStreamImpl classes) Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>

This is a follow up from apache#14659 that ports the windowed classes to use the StoreFactory abstraction as well. There's a side benefit of not duplicating the materialization code twice for each StreamImpl/CogroupedStreamImpl class as well. Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Matthias Sax <mjsax@apache.org>

refactor: introduce internal StoreFactory

c99f35e

ableegoldman changed the title ~~refactor: introduce internal StoreFactory~~ DO NOT MERGE: introduce internal StoreFactory Oct 31, 2023

ableegoldman reviewed Oct 31, 2023

View reviewed changes

agavra added 3 commits October 31, 2023 15:17

address feedback & have KVSMaterializer implement StoreFactory

0fee047

fix some things

90f9cf5

fix unit tests

dbb8010

agavra changed the title ~~DO NOT MERGE: introduce internal StoreFactory~~ KAFKA-15774: introduce internal StoreFactory Nov 2, 2023

agavra marked this pull request as ready for review November 2, 2023 23:59

ableegoldman reviewed Nov 3, 2023

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KeyValueStoreMaterializer.java Outdated Show resolved Hide resolved

ableegoldman reviewed Nov 3, 2023

View reviewed changes

address feedback

3a6dbdb

ableegoldman reviewed Nov 3, 2023

View reviewed changes

dont allow invalid StoreTypes

8110099

ableegoldman approved these changes Nov 7, 2023

View reviewed changes

ableegoldman merged commit febf0fb into apache:trunk Nov 7, 2023
1 check failed

agavra mentioned this pull request Nov 7, 2023

KAFKA-15774: refactor windowed stores to use StoreFactory #14708

Merged

3 tasks

mjsax reviewed Nov 8, 2023

View reviewed changes

agavra deleted the store_factory branch November 8, 2023 18:45

anurag-harness mentioned this pull request Feb 9, 2024

KAFKA-15774: refactor windowed stores to use StoreFactory (#14708) anurag-harness/kafka#233

Merged

mjsax added streams kip Requires or implements a KIP labels Feb 21, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15774: introduce internal StoreFactory #14659

KAFKA-15774: introduce internal StoreFactory #14659

agavra commented Oct 27, 2023 •

edited

ableegoldman Oct 31, 2023

agavra Oct 31, 2023

ableegoldman Oct 31, 2023

agavra Oct 31, 2023

ableegoldman Nov 1, 2023

ableegoldman Oct 31, 2023

agavra Oct 31, 2023 •

edited

ableegoldman left a comment

ableegoldman Nov 3, 2023

agavra Nov 3, 2023

ableegoldman Nov 3, 2023

agavra Nov 3, 2023

ableegoldman Nov 3, 2023

ableegoldman Nov 3, 2023

agavra Nov 3, 2023

agavra commented Nov 6, 2023

ableegoldman left a comment

mjsax Nov 8, 2023

KAFKA-15774: introduce internal StoreFactory #14659

KAFKA-15774: introduce internal StoreFactory #14659

Conversation

agavra commented Oct 27, 2023 • edited

Overview

Testing

Review Guide

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

agavra Oct 31, 2023 • edited

Choose a reason for hiding this comment

ableegoldman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

agavra commented Nov 6, 2023

ableegoldman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

agavra commented Oct 27, 2023 •

edited

agavra Oct 31, 2023 •

edited