I just updated the KIP to reflect the things discussed in this thread.
As for your questions Guozhang:
1) How do we handle if the num.partitions of app A's store changelog is
different from the num.tasks of app B's sub-topology with that read-only
store? Or are we going to let each task of B keep a whole copy of the
store
of A by reading all of its changelog partitions, like global stores?
Good question. Both need to be co-partitioned to have the data available.
Another option would be to use IQ to make the request, but that seems far
from ideal.
2) Are we trying to synchronize the store updates from the changelog to
app
B's processing timelines, or just like what we do for global stores that
we
just update the read-only stores async?
Pretty much the same as we do for global stores.
3) If the answer to both of the above questions are the latter, then
what's
the main difference of adding a read-only store v.s. adding a global
store?
I think because of the first answer the behavior differs from global
stores.
Makes sense?
Cheers,
D.
From: Matthias J. Sax <mj...@apache.org>
Date: Thursday, 20 January 2022 at 21:12
To: dev@kafka.apache.org <dev@kafka.apache.org>
Subject: Re: [DISCUSS] KIP-813 Shared State Stores
Any processor that would use that materialized, read-only statestore
would need to wait for the store to be restored. I can't find a way to make
that possible since processors can't wait for the statestore to be restored.
This is built into the runtime already. Nothing to worry about. It's
part of the regular restore logic -- as long as any store is restoring,
all processing is blocked.
Also, since the statestore would have logging disabled, it means there
is no initial restoration going on.
No. When we hookup the input topic as changelog (as the DSL does) we
restore from the input topic during regular restore phase. The restore
logic does not even know it's reading from the input topic, but not from
a "*-changelog" topic).
Disabling changelogging does only affect the write path (ie,
`store.put()`) but not the restore path due to the internal "hookup" of
the input topic inside the restore logic.
It's not easy to find/understand by reverse engineering I guess, but
it's there.
One pointer where the actual hookup happens (might help to dig into it
more if you want):
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/kstream/internals/InternalStreamsBuilder.java#L353-L356
-Matthias
On 1/20/22 10:04 AM, Guozhang Wang wrote:
Hello Daan,
Thanks for writing the KIP. I just read through it and just my 2c here:
to
me it seems that one of the goal would be to "externalize" the internal
changelog topic of an application (say A) so that other consumers can
directly read them --- though technically without any auth, anyone
knowing
the topic name would be able to write to it too, conceptually we would
just
assume that app A is the only writer of that topic --- The question I had
is how much we want to externalize the topic. For example we can,
orthogonally to this KIP, just allow users to pass in a customized topic
name when constructing a state store, indicating the application A to use
that as the changelog, and since that topic is created outside of A and
is
publicly visible to anyone else on that cluster, anyone --- including any
consumers, or streams apps. This is probably most flexible as for
sharing,
but we are even less assured that if application A is the only writer to
that external topic unless we have explicit auth for A on that topic.
Aside of that, here are a few more detailed comments about the
implementation design itself following your current proposal:
1) How do we handle if the num.partitions of app A's store changelog is
different from the num.tasks of app B's sub-topology with that read-only
store? Or are we going to let each task of B keep a whole copy of the
store
of A by reading all of its changelog partitions, like global stores?
2) Are we trying to synchronize the store updates from the changelog to
app
B's processing timelines, or just like what we do for global stores that
we
just update the read-only stores async?
3) If the answer to both of the above questions are the latter, then
what's
the main difference of adding a read-only store v.s. adding a global
store?
Guozhang
On Thu, Jan 20, 2022 at 6:27 AM Daan Gertis <dger...@korfinancial.com>
wrote:
Hey Matthias,
Thank you for that feedback, certainly some things to think about. Let
me
add my thoughts:
+1 on simplifying the motivation. Was aiming to add more context but I
think you're right, bringing it back to the essence makes more sense.
I also follow the reasoning of not having leader and follower. Makes
sense
to view it from a single app point of view.
As for the API method and its parameters, I wanted to stay close to the
API for adding a regular statestore, but I can perfectly find myself in
defining an addReadOnlyStateStore() method instead.
I agree the processor approach would be the most flexible one, and
surely
it allows you to use a processor to base the statestore off an existing
topic. From what I understood from the codebase, there might be a
problem
when using that statestore. Any processor that would use that
materialized,
read-only statestore would need to wait for the store to be restored. I
can't find a way to make that possible since processors can't wait for
the
statestore to be restored. Also, since the statestore would have logging
disabled, it means there is no initial restoration going on. As you
wrote,
the DSL is already doing this, so I'm pretty sure I'm missing something,
just unable to find what exactly.
I will rewrite the parts in the KIP to make processor-based the
preferred
choice, along with the changes to the motivation etc. Only thing to
figure
out is that restoring behavior to be sure processors of the readonly
statestore aren't working with stale data.
D.
-----Original Message-----
From: Matthias J. Sax <mj...@apache.org>
Sent: 19 January 2022 21:31
To: dev@kafka.apache.org
Subject: Re: [DISCUSS] KIP-813 Shared State Stores
Daan,
thanks for the KIP. I personally find the motivation section a little
bit
confusing. If I understand the KIP correctly, you want to read a topic
into
a state store (ie, materialize it). This is already possible today.
Of course, today a "second" changelog topic would be created. It seems
the
KIP aims to avoid the additional changelog topic, and to allow to re-use
the original input topic (this optimization is already available for the
DSL, but not for the PAPI).
If my observation is correct, we can simplify the motivation accordingly
(the fact that you want to use this feature to share state across
different
applications more efficiently seems to be secondary and we could omit it
IMHO to keep the motivation focused).
As a result, we also don't need to concept of "leader" and "follower".
In the end, Kafka Streams cannot reason/enforce any usage patterns
across
different apps, but we can only guarantee stuff within a single
application
(ie, don't create a changelog but reuse an input topic as changelog). It
would simplify the KIP if we remove these parts.
For the API, I am wondering why you propose to pass in `processorNames`?
To me, it seems more reasonable to pass a `ProcessorSupplier` instead
(similar to what we do for `addGlobalStore`)? The provided `Processor`
must
implement a certain pattern, ie, take each input record an apply it
unmodified to the state store (ie, the Processor will be solely
responsible
to maintain the state store). We might also need to pass in other
argument
similar to `addGlobalStore` into this method). (More below.)
If other processors need to read the state store, they can be connected
to
it explicitly via `connectProcessorAndStateStores()`? I guess a hybrid
approach to keep `processorName` would also be possible, but IMHO all
those
should only _read_ the state store (but not modify it), to keep a clear
conceptual separation.
About the method name: wondering if we should use a different name to be
more explicit what the method does? Maybe `addReadOnlyStateStore`?
Btw: please omit any code snippets and only put the newly added method
signature in the KIP.
What I don't yet understand is the section "Allow state stores to
continue listening for changes from their changelog". Can you elaborate?
About:
Since a changelog topic is created with the application id in it’s
name,
it would allow us to check in the follower if the changelog topic starts
with our application id. If it doesn’t, we are not allowed to send a
log.
The DSL implements this differently, and just disabled the changelog for
the state store (ie, for the "follower"). We could do the same thing
(either enforcing that the provided `StoreBuilder` has changelogging
disabled, or by just ignoring it and disabled it hard coded).
Ie, overall I would prefer the "source-procssor appraoch" that you put
into rejected alternatives. Note that the problem you call out, namely
Problem with this approach is the lack of having restoring support
within the state store
does not apply. A restore it absolutely possible and the DSL already
supports it.
Or is your concern with regard to performance? The "source-processor
approach" would have the disadvantage that input data is first
deserialized, fed into the Processor, and than serialized again when put
into the state store. Re-using the state restore code is a good idea
from a performance point of view, but it might require quite some
internal changes (your proposal to "not stop restoring" might not work
as it could trigger quite some undesired side-effects given the current
architecture of Kafka Streams).
-Matthias
On 1/16/22 11:52 PM, Daan Gertis wrote:
Hey everyone,
Just created a KIP on sharing statestore state across multiple
applications without duplicating the data on multiple changelog topics.
Have a look and tell me what you think or what to improve. This is my
first
one, so please be gentle 😉
https://cwiki.apache.org/confluence/x/q53kCw
Cheers!
D.
