Hey Nick,
Thank you for the prototype testing and benchmarking, and sorry for the
late reply!
I agree that it is worth revisiting the WriteBatchWithIndex approach. I
will implement a fork of the current prototype that uses that mechanism to
ensure transactionality and let you know when it is ready for
review/testing in this ML thread.
As for time estimates, I might not have enough time to finish the prototype
in December, so it will probably be ready for review in January.
Best,
Alex
On Fri, Nov 11, 2022 at 4:24 PM Nick Telford <nick.telf...@gmail.com>
wrote:
Hi everyone,
Sorry to dredge this up again. I've had a chance to start doing some
testing with the WIP Pull Request, and it appears as though the secondary
store solution performs rather poorly.
In our testing, we had a non-transactional state store that would restore
(from scratch), at a rate of nearly 1,000,000 records/second. When we
switched it to a transactional store, it restored at a rate of less than
40,000 records/second.
I suspect the key issues here are having to copy the data out of the
temporary store and into the main store on-commit, and to a lesser
extent,
the extra memory copies during writes.
I think it's worth re-visiting the WriteBatchWithIndex solution, as it's
clear from the RocksDB post[1] on the subject that it's the recommended
way
to achieve transactionality.
The only issue you identified with this solution was that uncommitted
writes are required to entirely fit in-memory, and RocksDB recommends
they
don't exceed 3-4MiB. If we do some back-of-the-envelope calculations, I
think we'll find that this will be a non-issue for all but the most
extreme
cases, and for those, I think I have a fairly simple solution.
Firstly, when EOS is enabled, the default commit.interval.ms is set to
100ms, which provides fairly short intervals that uncommitted writes need
to be buffered in-memory. If we assume a worst case of 1024 byte records
(and for most cases, they should be much smaller), then 4MiB would hold
~4096 records, which with 100ms commit intervals is a throughput of
approximately 40,960 records/second. This seems quite reasonable.
For use cases that wouldn't reasonably fit in-memory, my suggestion is
that
we have a mechanism that tracks the number/size of uncommitted records in
stores, and prematurely commits the Task when this size exceeds a
configured threshold.
Thanks for your time, and let me know what you think!
--
Nick
1: https://rocksdb.org/blog/2015/02/27/write-batch-with-index.html
On Thu, 6 Oct 2022 at 19:31, Alexander Sorokoumov
<asorokou...@confluent.io.invalid> wrote:
Hey Nick,
It is going to be option c. Existing state is considered to be
committed
and there will be an additional RocksDB for uncommitted writes.
I am out of office until October 24. I will update KIP and make sure
that
we have an upgrade test for that after coming back from vacation.
Best,
Alex
On Thu, Oct 6, 2022 at 5:06 PM Nick Telford <nick.telf...@gmail.com>
wrote:
Hi everyone,
I realise this has already been voted on and accepted, but it
occurred
to
me today that the KIP doesn't define the migration/upgrade path for
existing non-transactional StateStores that *become* transactional,
i.e.
by
adding the transactional boolean to the StateStore constructor.
What would be the result, when such a change is made to a Topology,
without
explicitly wiping the application state?
a) An error.
b) Local state is wiped.
c) Existing RocksDB database is used as committed writes and new
RocksDB
database is created for uncommitted writes.
d) Something else?
Regards,
Nick
On Thu, 1 Sept 2022 at 12:16, Alexander Sorokoumov
<asorokou...@confluent.io.invalid> wrote:
Hey Guozhang,
Sounds good. I annotated all added StateStore methods (commit,
recover,
transactional) with @Evolving.
Best,
Alex
On Wed, Aug 31, 2022 at 7:32 PM Guozhang Wang <wangg...@gmail.com>
wrote:
Hello Alex,
Thanks for the detailed replies, I think that makes sense, and in
the
long
run we would need some public indicators from StateStore to
determine
if
checkpoints can really be used to indicate clean snapshots.
As for the @Evolving label, I think we can still keep it but for
a
different reason, since as we add more state management
functionalities
in
the near future we may need to revisit the public APIs again and
hence
keeping it as @Evolving would allow us to modify if necessary, in
an
easier
path than deprecate -> delete after several minor releases.
Besides that, I have no further comments about the KIP.
Guozhang
On Fri, Aug 26, 2022 at 1:51 AM Alexander Sorokoumov
<asorokou...@confluent.io.invalid> wrote:
Hey Guozhang,
I think that we will have to keep StateStore#transactional()
because
post-commit checkpointing of non-txn state stores will break
the
guarantees
we want in
ProcessorStateManager#initializeStoreOffsetsFromCheckpoint
for
correct recovery. Let's consider checkpoint-recovery behavior
under
EOS
that we want to support:
1. Non-txn state stores should checkpoint on graceful shutdown
and
restore
from that checkpoint.
2. Non-txn state stores should delete local data during
recovery
after
a
crash failure.
3. Txn state stores should checkpoint on commit and on graceful
shutdown.
These stores should roll back uncommitted changes instead of
deleting
all
local data.
#1 and #2 are already supported; this proposal adds #3.
Essentially,
we
have two parties at play here - the post-commit checkpointing
in
StreamTask#postCommit and recovery in ProcessorStateManager#
initializeStoreOffsetsFromCheckpoint. Together, these methods
must
allow
all three workflows and prevent invalid behavior, e.g., non-txn
stores
should not checkpoint post-commit to avoid keeping uncommitted
data
on
recovery.
In the current state of the prototype, we checkpoint only txn
state
stores
post-commit under EOS using StateStore#transactional(). If we
remove
StateStore#transactional() and always checkpoint post-commit,
ProcessorStateManager#initializeStoreOffsetsFromCheckpoint will
have
to
determine whether to delete local data. Non-txn implementation
of
StateStore#recover can't detect if it has uncommitted writes.
Since
its
default implementation must always return either true or false,
signaling
whether it is restored into a valid committed-only state. If
StateStore#recover always returns true, we preserve uncommitted
writes
and
violate correctness. Otherwise, ProcessorStateManager#
initializeStoreOffsetsFromCheckpoint would always delete local
data
even
after
a graceful shutdown.
With StateStore#transactional we avoid checkpointing non-txn
state
stores
and prevent that problem during recovery.
Best,
Alex
On Fri, Aug 19, 2022 at 1:05 AM Guozhang Wang <
wangg...@gmail.com>
wrote:
Hello Alex,
Thanks for the replies!
As long as we allow custom user implementations of that
interface,
we
should
probably either keep that flag to distinguish between
transactional
and
non-transactional implementations or change the contract
behind
the
interface. What do you think?
Regarding this question, I thought that in the long run, we
may
always
write checkpoints regardless of txn v.s. non-txn stores, in
which
case
we
would not need that `StateStore#transactional()`. But for now
in
order
for
backward compatibility edge cases we still need to
distinguish
on
whether
or not to write checkpoints. Maybe I was mis-reading its
purposes?
If
yes,
please let me know.
On Mon, Aug 15, 2022 at 7:56 AM Alexander Sorokoumov
<asorokou...@confluent.io.invalid> wrote:
Hey Guozhang,
Thank you for elaborating! I like your idea to introduce a
StreamsConfig
specifically for the default store APIs. You mentioned
Materialized,
but
I
think changes in StreamJoined follow the same logic.
I updated the KIP and the prototype according to your
suggestions:
* Add a new StoreType and a StreamsConfig for transactional
RocksDB.
* Decide whether Materialized/StreamJoined are
transactional
based
on
the
configured StoreType.
* Move RocksDBTransactionalMechanism to
org.apache.kafka.streams.state.internals to remove it from
the
proposal
scope.
* Add a flag in new Stores methods to configure a state
store
as
transactional. Transactional state stores use the default
transactional
mechanism.
* The changes above allowed to remove all changes to the
StoreSupplier
interface.
I am not sure about marking StateStore#transactional() as
evolving.
As
long
as we allow custom user implementations of that interface,
we
should
probably either keep that flag to distinguish between
transactional
and
non-transactional implementations or change the contract
behind
the
interface. What do you think?
Best,
Alex
On Thu, Aug 11, 2022 at 1:00 AM Guozhang Wang <
wangg...@gmail.com>
wrote:
Hello Alex,
Thanks for the replies. Regarding the global config v.s.
per-store
spec,
I
agree with John's early comments to some degrees, but I
think
we
may
well
distinguish a couple scenarios here. In sum we are
discussing
about
the
following levels of per-store spec:
* Materialized#transactional()
* StoreSupplier#transactional()
* StateStore#transactional()
* Stores.persistentTransactionalKeyValueStore()...
And my thoughts are the following:
* In the current proposal users could specify
transactional
as
either
"Materialized.as("storeName").withTransantionsEnabled()"
or
"Materialized.as(Stores.persistentTransactionalKeyValueStore(..))",
which
seems not necessary to me. In general, the more options
the
library
provides, the messier for users to learn the new APIs.
* When using built-in stores, users would usually go with
Materialized.as("storeName"). In such cases I feel it's
not
very
meaningful
to specify "some of the built-in stores to be
transactional,
while
others
be non transactional": as long as one of your stores are
non-transactional,
you'd still pay for large restoration cost upon unclean
failure.
People
may, indeed, want to specify if different transactional
mechanisms
to
be
used across stores; but for whether or not the stores
should
be
transactional, I feel it's really an "all or none"
answer,
and
our
built-in
form (rocksDB) should support transactionality for all
store
types.
* When using customized stores, users would usually go
with
Materialized.as(StoreSupplier). And it's possible if
users
would
choose
some to be transactional while others non-transactional
(e.g.
if
their
customized store only supports transactional for some
store
types,
but
not
others).
* At a per-store level, the library do not really care,
or
need
to
know
whether that store is transactional or not at runtime,
except
for
compatibility reasons today we want to make sure the
written
checkpoint
files do not include those non-transactional stores. But
this
check
would
eventually go away as one day we would always checkpoint
files.
---------------------------
With all of that in mind, my gut feeling is that:
* Materialized#transactional(): we would not need this
knob,
since
for
built-in stores I think just a global config should be
sufficient
(see
below), while for customized store users would need to
specify
that
via
the
StoreSupplier anyways and not through this API. Hence I
think
for
either
case we do not need to expose such a knob on the
Materialized
level.
* Stores.persistentTransactionalKeyValueStore(): I think
we
could
refactor
that function without introducing new constructors in the
Stores
factory,
but just add new overloads to the existing func name e.g.
```
persistentKeyValueStore(final String name, final boolean
transactional)
```
Plus we can augment the storeImplType as introduced in
https://cwiki.apache.org/confluence/display/KAFKA/KIP-591%3A+Add+Kafka+Streams+config+to+set+default+state+store
as a syntax sugar for users, e.g.
```
public enum StoreImplType {
ROCKS_DB,
TXN_ROCKS_DB,
IN_MEMORY
}
```
```
stream.groupByKey().count(Materialized.withStoreType(StoreImplType.TXN_
ROCKS_DB));
```
The above provides this global config at the store impl
type
level.
* RocksDBTransactionalMechanism: I agree with Bruno that
we
would
better
not expose this knob to users, but rather keep it purely
as
an
impl
detail
abstracted from the "TXN_ROCKS_DB" type. Over time we
may,
e.g.
use
in-memory stores as the secondary stores with optional
spill-to-disks
when
we hit the memory limit, but all of that optimizations in
the
future
should
be kept away from the users.
* StoreSupplier#transactional() /
StateStore#transactional():
the
first
flag is only used to be passed into the StateStore layer,
for
indicating
if
we should write checkpoints; we could mark it as
@evolving
so
that
we
can
one day remove it without a long deprecation period.
Guozhang
On Wed, Aug 10, 2022 at 8:04 AM Alexander Sorokoumov
<asorokou...@confluent.io.invalid> wrote:
Hey Guozhang, Bruno,
Thank you for your feedback. I am going to respond to
both
of
you
in
a
single email. I hope it is okay.
@Guozhang,
We could, instead, have a global
config to specify if the built-in stores should be
transactional
or
not.
This was the original approach I took in this proposal.
Earlier
in
this
thread John, Sagar, and Bruno listed a number of issues
with
it.
I
tend
to
agree with them that it is probably better user
experience
to
control
transactionality via Materialized objects.
We could simplify our implementation for `commit`
Agreed! I updated the prototype and removed references
to
the
commit
marker
and rolling forward from the proposal.
@Bruno,
So, I would remove the details about the 2-state-store
implementation
from the KIP or provide it as an example of a
possible
implementation
at
the end of the KIP.
I moved the section about the 2-state-store
implementation
to
the
bottom
of
the proposal and always mention it as a reference
implementation.
Please
let me know if this is okay.
Could you please describe the usage of commit() and
recover()
in
the
commit workflow in the KIP as we did in this thread
but
independently
from the state store implementation?
I described how commit/recover change the workflow in
the
Overview
section.
Best,
Alex
On Wed, Aug 10, 2022 at 10:07 AM Bruno Cadonna <
cado...@apache.org
wrote:
Hi Alex,
Thank a lot for explaining!
Now some aspects are clearer to me.
While I understand now, how the state store can roll
forward, I
have
the
feeling that rolling forward is specific to the
2-state-store
implementation with RocksDB of your PoC. Other state
store
implementations might use a different strategy to
react
to
crashes.
For
example, they might apply an atomic write and
effectively
rollback
if
they crash before committing the state store
transaction. I
think
the
KIP should not contain such implementation details
but
provide
an
interface to accommodate rolling forward and rolling
backward.
So, I would remove the details about the
2-state-store
implementation
from the KIP or provide it as an example of a
possible
implementation
at
the end of the KIP.
Since a state store implementation can roll forward
or
roll
back, I
think it is fine to return the changelog offset from
recover().
With
the
returned changelog offset, Streams knows from where
to
start
state
store
restoration.
Could you please describe the usage of commit() and
recover()
in
the
commit workflow in the KIP as we did in this thread
but
independently
from the state store implementation? That would make
things
clearer.
Additionally, descriptions of failure scenarios would
also
be
helpful.
Best,
Bruno
On 04.08.22 16:39, Alexander Sorokoumov wrote:
Hey Bruno,
Thank you for the suggestions and the clarifying
questions. I
believe
that
they cover the core of this proposal, so it is
crucial
for
us
to
be
on
the
same page.
1. Don't you want to deprecate StateStore#flush().
Good call! I updated both the proposal and the
prototype.
2. I would shorten
Materialized#withTransactionalityEnabled()
to
Materialized#withTransactionsEnabled().
Turns out, these methods are no longer necessary. I
removed
them
from
the
proposal and the prototype.
3. Could you also describe a bit more in detail
where
the
offsets
passed
into commit() and recover() come from?
The offset passed into StateStore#commit is the
last
offset
committed
to
the changelog topic. The offset passed into
StateStore#recover
is
the
last
checkpointed offset for the given StateStore. Let's
look
at
steps 3
and 4
in the commit workflow. After the
TaskExecutor/TaskManager
commits,
it
calls
StreamTask#postCommit[1] that in turn:
a. updates the changelog offsets via
ProcessorStateManager#updateChangelogOffsets[2].
The
offsets
here
come
from
the RecordCollector[3], which tracks the latest
offsets
the
producer
sent
without exception[4, 5].
b. flushes/commits the state store in
AbstractTask#maybeCheckpoint[6].
This
method essentially calls ProcessorStateManager
methods
-
flush/commit[7]
and checkpoint[8]. ProcessorStateManager#commit
goes
over
all
state
stores
that belong to that task and commits them with the
offset
obtained
in
step
`a`. ProcessorStateManager#checkpoint writes down
those
offsets
for
all
state stores, except for non-transactional ones in
the
case
of
EOS.
During initialization, StreamTask calls
StateManagerUtil#registerStateStores[8] that in
turn
calls
ProcessorStateManager#initializeStoreOffsetsFromCheckpoint[9].
At
the
moment, this method assigns checkpointed offsets to
the
corresponding
state
stores[10]. The prototype also calls
StateStore#recover
with
the
checkpointed offset and assigns the offset returned
by
recover()[11].
4. I do not quite understand how a state store can
roll
forward.
You
mention in the thread the following:
The 2-state-stores commit looks like this [12]:
1. Flush the temporary state store.
2. Create a commit marker with a changelog
offset
corresponding
to
the
state we are committing.
3. Go over all keys in the temporary store and
write
them
down
to
the
main one.
4. Wipe the temporary store.
5. Delete the commit marker.
Let's consider crash failure scenarios:
- Crash failure happens between steps 1 and 2.
The
main
state
store
is
in a consistent state that corresponds to the
previously
checkpointed
offset. StateStore#recover throws away the
temporary
store
and
proceeds
from the last checkpointed offset.
- Crash failure happens between steps 2 and 3.
We
do
not
know
what
keys
from the temporary store were already written
to
the
main
store,
so
we
can't roll back. There are two options - either
wipe
the
main
store
or roll
forward. Since the point of this proposal is to
avoid
situations
where we
throw away the state and we do not care to what
consistent
state
the
store
rolls to, we roll forward by continuing from
step
3.
- Crash failure happens between steps 3 and 4.
We
can't
distinguish
between this and the previous scenario, so we
write
all
the
keys
from the
temporary store. This is okay because the
operation
is
idempotent.
- Crash failure happens between steps 4 and 5.
Again,
we
can't
distinguish between this and previous
scenarios,
but
the
temporary
store is
already empty. Even though we write all keys
from
the
temporary
store, this
operation is, in fact, no-op.
- Crash failure happens between step 5 and
checkpoint.
This
is
the
case
you referred to in question 5. The commit is
finished,
but
it
is
not
reflected at the checkpoint. recover() returns
the
offset
of
the
previous
commit here, which is incorrect, but it is okay
because
we
will
replay the
changelog from the previously committed offset.
As
changelog
replay
is
idempotent, the state store recovers into a
consistent
state.
The last crash failure scenario is a natural
transition
to
how should Streams know what to write into the
checkpoint
file
after the crash?
As mentioned above, the Streams app writes the
checkpoint
file
after
the
Kafka transaction and then the StateStore commit.
Same
as
without
the
proposal, it should write the committed offset, as
it
is
the
same
for
both
the Kafka changelog and the state store.
This issue arises because we store the offset
outside
of
the
state
store. Maybe we need an additional method on the
state
store
interface
that returns the offset at which the state store
is.
In my opinion, we should include in the interface
only
the
guarantees
that
are necessary to preserve EOS without wiping the
local
state.
This
way,
we
allow more room for possible implementations.
Thanks
to
the
idempotency
of
the changelog replay, it is "good enough" if
StateStore#recover
returns
the
offset that is less than what it actually is. The
only
limitation
here
is
that the state store should never commit writes
that
are
not
yet
committed
in Kafka changelog.
Please let me know what you think about this. First
of
all, I
am
relatively
new to the codebase, so I might be wrong in my
understanding
of
how it works. Second, while writing this, it
occured
to
me
that
the
StateStore#recover interface method is not
straightforward
as
it
can
be.
Maybe we can change it like that:
/**
* Recover a transactional state store
* <p>
* If a transactional state store shut down
with
a
crash
failure,
this
method ensures that the
* state store is in a consistent state that
corresponds
to
{@code
changelofOffset} or later.
*
* @param changelogOffset the checkpointed
changelog
offset.
* @return {@code true} if recovery succeeded,
{@code
false}
otherwise.
*/
boolean recover(final Long changelogOffset) {
Note: all links below except for [10] lead to the
prototype's
code.
1.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L468
2.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L580
3.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L868
4.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L94-L96
5.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java#L213-L216
6.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java#L94-L97
7.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L469
8.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L226
9.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateManagerUtil.java#L103
10.
https://github.com/apache/kafka/blob/0c4da23098f8b8ae9542acd7fbaa1e5c16384a39/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L251-L252
11.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java#L250-L265
12.
https://github.com/apache/kafka/blob/549e54be95a8e1bae1e97df2c21d48c042ff356e/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractTransactionalStore.java#L84-L88
Best,
Alex
On Fri, Jul 29, 2022 at 3:42 PM Bruno Cadonna <
cado...@apache.org>
wrote:
Hi Alex,
Thanks for the updates!
1. Don't you want to deprecate StateStore#flush().
As
far
as I
understand, commit() is the new flush(), right? If
you
do
not
deprecate
it, you don't get rid of the error room you
describe
in
your
KIP
by
having a flush() and a commit().
2. I would shorten
Materialized#withTransactionalityEnabled()
to
Materialized#withTransactionsEnabled().
3. Could you also describe a bit more in detail
where
the
offsets
passed
into commit() and recover() come from?
For my next two points, I need the commit workflow
that
you
were
so
kind
to post into this thread:
1. write stuff to the state store
2. producer.sendOffsetsToTransaction(token);
producer.commitTransaction();
3. flush (<- that would be call to commit(),
right?)
4. checkpoint
4. I do not quite understand how a state store can
roll
forward.
You
mention in the thread the following:
"If the crash failure happens during #3, the state
store
can
roll
forward and finish the flush/commit."
How does the state store know where it stopped the
flushing
when
it
crashed?
This seems an optimization to me. I think in
general
the
state
store
should rollback to the last successfully committed
state
and
restore
from there until the end of the changelog topic
partition.
The
last
committed state is the offsets in the checkpoint
file.
5. In the same e-mail from point 4, you also
state:
"If the crash failure happens between #3 and #4,
the
state
store
should
do nothing during recovery and just proceed with
the
checkpoint."
How should Streams know that the failure was
between
#3
and
#4
during
recovery? It just sees a valid state store and a
valid
checkpoint
file.
Streams does not know that the state of the
checkpoint
file
does
not
match with the committed state of the state store.
Also, how should Streams know what to write into
the
checkpoint
file
after the crash?
This issue arises because we store the offset
outside
of
the
state
store. Maybe we need an additional method on the
state
store
interface
that returns the offset at which the state store
is.
Best,
Bruno
On 27.07.22 11:51, Alexander Sorokoumov wrote:
Hey Nick,
Thank you for the kind words and the feedback!
I'll
definitely
add
an
option to configure the transactional mechanism
in
Stores
factory
method
via an argument as John previously suggested and
might
add
the
in-memory
option via RocksDB Indexed Batches if I figure
why
their
creation
via
rocksdb jni fails with
`UnsatisfiedLinkException`.
Best,
Alex
On Wed, Jul 27, 2022 at 11:46 AM Alexander
Sorokoumov <
asorokou...@confluent.io> wrote:
Hey Guozhang,
1) About the param passed into the `recover()`
function:
it
seems
to
me
that the semantics of "recover(offset)" is:
recover
this
state
to a
transaction boundary which is at least the
passed-in
offset.
And
the
only
possibility that the returned offset is
different
than
the
passed-in
offset
is that if the previous failure happens after
we've
done
all
the
commit
procedures except writing the new checkpoint,
in
which
case
the
returned
offset would be larger than the passed-in
offset.
Otherwise
it
should
always be equal to the passed-in offset, is
that
right?
Right now, the only case when `recover` returns
an
offset
different
from
the passed one is when the failure happens
*during*
commit.
If the failure happens after commit but before
the
checkpoint,
`recover`
might return either a passed or newer committed
offset,
depending
on
the
implementation. The `recover` implementation in
the
prototype
returns
a
passed offset because it deletes the commit
marker
that
holds
that
offset
after the commit is done. In that case, the
store
will
replay
the
last
commit from the changelog. I think it is fine as
the
changelog
replay
is
idempotent.
2) It seems the only use for the
"transactional()"
function
is
to
determine
if we can update the checkpoint file while in
EOS.
Right now, there are 2 other uses for
`transactional()`:
1. To determine what to do during initialization
if
the
checkpoint
is
gone
(see [1]). If the state store is transactional,
we
don't
have
to
wipe
the
existing data. Thinking about it now, we do not
really
need
this
check
whether the store is `transactional` because if
it
is
not,
we'd
not
have
written the checkpoint in the first place. I am
going
to
remove
that
check.
2. To determine if the persistent kv store in
KStreamImplJoin
should
be
transactional (see [2], [3]).
I am not sure if we can get rid of the checks in
point
2.
If
so,
I'd
be
happy to encapsulate `transactional()` logic in
`commit/recover`.
Best,
Alex
1.
https://github.com/apache/kafka/pull/12393/files#diff-971d9ef7ea8aefffff687fc7ee131bd166ced94445f4ab55aa83007541dccfdaL256-R281
2.
https://github.com/apache/kafka/pull/12393/files#diff-9ce43046fdef1233ab762e728abd1d3d44d7c270b28dcf6b63aa31a93a30af07R266-R278
3.
https://github.com/apache/kafka/pull/12393/files#diff-9ce43046fdef1233ab762e728abd1d3d44d7c270b28dcf6b63aa31a93a30af07R348-R354
On Tue, Jul 26, 2022 at 6:39 PM Nick Telford <
nick.telf...@gmail.com>
wrote:
Hi Alex,
Excellent proposal, I'm very keen to see this
land!
Would it be useful to permit configuring the
type
of
store
used
for
uncommitted offsets on a store-by-store basis?
This
way,
users
could
choose
whether to use, e.g. an in-memory store or
RocksDB,
potentially
reducing
the overheads associated with RocksDb for
smaller
stores,
but
without
the
memory pressure issues?
I suspect that in most cases, the number of
uncommitted
records
will
be
very small, because the default commit interval
is
100ms.
Regards,
Nick
On Tue, 26 Jul 2022 at 01:36, Guozhang Wang <
wangg...@gmail.com>
wrote:
Hello Alex,
Thanks for the updated KIP, I looked over it
and
browsed
the
WIP
and
just
have a couple meta thoughts:
1) About the param passed into the `recover()`
function:
it
seems
to
me
that the semantics of "recover(offset)" is:
recover
this
state
to
a
transaction boundary which is at least the
passed-in
offset.
And
the
only
possibility that the returned offset is
different
than
the
passed-in
offset
is that if the previous failure happens after
we've
done
all
the
commit
procedures except writing the new checkpoint,
in
which
case
the
returned
offset would be larger than the passed-in
offset.
Otherwise
it
should
always be equal to the passed-in offset, is
that
right?
2) It seems the only use for the
"transactional()"
function
is
to
determine
if we can update the checkpoint file while in
EOS.
But
the
purpose
of
the
checkpoint file's offsets is just to tell "the
local
state's
current
snapshot's progress is at least the indicated
offsets"
anyways,
and
with
this KIP maybe we would just do:
a) when in ALOS, upon failover: we set the
starting
offset
as
checkpointed-offset, then restore() from
changelog
till
the
end-offset.
This way we may restore some records twice.
b) when in EOS, upon failover: we first call
recover(checkpointed-offset),
then set the starting offset as the returned
offset
(which
may
be
larger
than checkpointed-offset), then restore until
the
end-offset.
So why not also:
c) we let the `commit()` function to also
return
an
offset,
which
indicates
"checkpointable offsets".
d) for existing non-transactional stores, we
just
have a
default
implementation of "commit()" which is simply a
flush,
and
returns
a
sentinel value like -1. Then later if we get
checkpointable
offsets
-1,
we
do not write the checkpoint. Upon clean
shutting
down
we
can
just
checkpoint regardless of the returned value
from
"commit".
e) for existing non-transactional stores, we
just
have a
default
implementation of "recover()" which is to wipe
out
the
local
store
and
return offset 0 if the passed in offset is -1,
otherwise
if
not
-1
then
it
indicates a clean shutdown in the last run,
can
this
function
is
just
a
no-op.
In that case, we would not need the
"transactional()"
function
anymore,
since for non-transactional stores their
behaviors
are
still
wrapped
in
the
`commit / recover` function pairs.
I have not completed the thorough pass on your
WIP
PR,
so
maybe
I
could
come up with some more feedback later, but
just
let
me
know
if
my
understanding above is correct or not?
Guozhang
On Thu, Jul 14, 2022 at 7:01 AM Alexander
Sorokoumov
<asorokou...@confluent.io.invalid> wrote:
Hi,
I updated the KIP with the following changes:
* Replaced in-memory batches with the
secondary-store
approach
as
the
default implementation to address the
feedback
about
memory
pressure
as
suggested by Sagar and Bruno.
* Introduced StateStore#commit and
StateStore#recover
methods
as
an
extension of the rollback idea. @Guozhang,
please
see
the
comment
below
on
why I took a slightly different approach than
you
suggested.
* Removed mentions of changes to IQv1 and
IQv2.
Transactional
state
stores
enable reading committed in IQ, but it is
really
an
independent
feature
that deserves its own KIP. Conflating them
unnecessarily
increases
the
scope for discussion, implementation, and
testing
in
a
single
unit
of
work.
I also published a prototype -
https://github.com/apache/kafka/pull/12393
that implements changes described in the
proposal.
Regarding explicit rollback, I think it is a
powerful
idea
that
allows
other StateStore implementations to take a
different
path
to
the
transactional behavior rather than keep 2
state
stores.
Instead
of
introducing a new commit token, I suggest
using a
changelog
offset
that
already 1:1 corresponds to the materialized
state.
This
works
nicely
because Kafka Stream first commits an AK
transaction
and
only
then
checkpoints the state store, so we can use
the
changelog
offset
to
commit
the state store transaction.
I called the method StateStore#recover rather
than
StateStore#rollback
because a state store might either roll back
or
forward
depending
on
the
specific point of the crash failure.Consider
the
write
algorithm
in
Kafka
Streams is:
1. write stuff to the state store
2. producer.sendOffsetsToTransaction(token);
producer.commitTransaction();
3. flush
4. checkpoint
Let's consider 3 cases:
1. If the crash failure happens between #2
and
#3,
the
state
store
rolls
back and replays the uncommitted transaction
from
the
changelog.
2. If the crash failure happens during #3,
the
state
store
can
roll
forward
and finish the flush/commit.
3. If the crash failure happens between #3
and
#4,
the
state
store
should
do nothing during recovery and just proceed
with
the
checkpoint.
Looking forward to your feedback,
Alexander
On Wed, Jun 8, 2022 at 12:16 AM Alexander
Sorokoumov
<
asorokou...@confluent.io> wrote:
Hi,
As a status update, I did the following
changes
to
the
KIP:
* replaced configuration via the top-level
config
with
configuration
via
Stores factory and StoreSuppliers,
* added IQv2 and elaborated how
readCommitted
will
work
when
the
store
is
not transactional,
* removed claims about ALOS.
I am going to be OOO in the next couple of
weeks
and
will
resume
working
on the proposal and responding to the
discussion
in
this
thread
starting
June 27. My next top priorities are:
1. Prototype the rollback approach as
suggested
by
Guozhang.
2. Replace in-memory batches with the
secondary-store
approach
as
the
default implementation to address the
feedback
about
memory
pressure as
suggested by Sagar and Bruno.
3. Adjust Stores methods to make
transactional
implementations
pluggable.
4. Publish the POC for the first review.
Best regards,
Alex
On Wed, Jun 1, 2022 at 2:52 PM Guozhang
Wang <
wangg...@gmail.com>
wrote:
Alex,
Thanks for your replies! That is very
helpful.
Just to broaden our discussions a bit
here, I
think
there
are
some
other
approaches in parallel to the idea of
"enforce
to
only
persist
upon
explicit flush" and I'd like to throw one
here
--
not
really
advocating
it,
but just for us to compare the pros and
cons:
1) We let the StateStore's `flush` function
to
return a
token
instead
of
returning `void`.
2) We add another `rollback(token)`
interface
of
StateStore
which
would
effectively rollback the state as indicated
by
the
token
to
the
snapshot
when the corresponding `flush` is called.
3) We encode the token and commit as part
of
`producer#sendOffsetsToTransaction`.
Users could optionally implement the new
functions,
or
they
can
just
not
return the token at all and not implement
the
second
function.
Again,
the
APIs are just for the sake of illustration,
not
feeling
they
are
the
most
natural :)
Then the procedure would be:
1. the previous checkpointed offset is 100
...
3. flush store, make sure all writes are
persisted;
get
the
returned
token
that indicates the snapshot of 200.
4.
producer.sendOffsetsToTransaction(token);
producer.commitTransaction();
5. Update the checkpoint file (say, the new
value
is
200).
Then if there's a failure, say between 3/4,
we
would
get
the
token
from
the
last committed txn, and first we would do
the
restoration
(which
may
get
the state to somewhere between 100 and
200),
then
call
`store.rollback(token)` to rollback to the
snapshot
of
offset
100.
The pros is that we would then not need to
enforce
the
state
stores to
not
persist any data during the txn: for stores
that
may
not
be
able
to
implement the `rollback` function, they can
still
reduce
its
impl
to
"not
persisting any data" via this API, but for
stores
that
can
indeed
support
the rollback, their implementation may be
more
efficient.
The
cons
though,
on top of my head are 1) more complicated
logic
differentiating
between
EOS
with and without store rollback support,
and
ALOS,
2)
encoding
the
token
as
part of the commit offset is not ideal if
it
is
big,
3)
the
recovery
logic
including the state store is also a bit
more
complicated.
Guozhang
On Wed, Jun 1, 2022 at 1:29 PM Alexander
Sorokoumov
<asorokou...@confluent.io.invalid> wrote:
Hi Guozhang,
But I'm still trying to clarify how it
guarantees
EOS,
and
it
seems
that we
would achieve it by enforcing to not
persist
any
data
written
within
this
transaction until step 4. Is that
correct?
This is correct. Both alternatives -
in-memory
WriteBatchWithIndex
and
transactionality via the secondary store
guarantee
EOS
by
not
persisting
data in the "main" state store until it is
committed
in
the
changelog
topic.
Oh what I meant is not what KStream code
does,
but
that
StateStore
impl
classes themselves could potentially
flush
data
to
become
persisted
asynchronously
Thank you for elaborating! You are
correct,
the
underlying
state
store
should not persist data until the streams
app
calls
StateStore#flush.
There
are 2 options how a State Store
implementation
can
guarantee
that -
either
keep uncommitted writes in memory or be
able
to
roll
back
the
changes
that
were not committed during recovery.
RocksDB's
WriteBatchWithIndex is
an
implementation of the first option. A
considered
alternative,
Transactions
via Secondary State Store for Uncommitted
Changes,
is
the
way
to
implement
the second option.
As everyone correctly pointed out, keeping
uncommitted
data
in
memory
introduces a very real risk of OOM that we
will
need
to
handle.
The
more I
think about it, the more I lean towards
going
with
the
Transactions
via
Secondary Store as the way to implement
transactionality
as
it
does
not
have that issue.
Best,
Alex
On Wed, Jun 1, 2022 at 12:59 PM Guozhang
Wang
<
wangg...@gmail.com>
wrote:
Hello Alex,
we flush the cache, but not the
underlying
state
store.
You're right. The ordering I mentioned
above
is
actually:
...
3. producer.sendOffsetsToTransaction();
producer.commitTransaction();
4. flush store, make sure all writes are
persisted.
5. Update the checkpoint file to 200.
But I'm still trying to clarify how it
guarantees
EOS,
and
it
seems
that
we
would achieve it by enforcing to not
persist
any
data
written
within
this
transaction until step 4. Is that
correct?
Can you please point me to the place in
the
codebase
where
we
trigger
async flush before the commit?
Oh what I meant is not what KStream code
does,
but
that
StateStore
impl
classes themselves could potentially
flush
data
to
become
persisted
asynchronously, e.g. RocksDB does that
naturally
out
of
the
control
of
KStream code. I think it is related to my
previous
question:
if we
think
by
guaranteeing EOS at the state store
level,
we
would
effectively
ask
the
impl classes that "you should not persist
any
data
until
`flush`
is
called
explicitly", is the StateStore interface
the
right
level
to
enforce
such
mechanisms, or should we just do that on
top
of
the
StateStores,
e.g.
during the transaction we just keep all
the
writes
in
the
cache
(of
course
we need to consider how to work around
memory
pressure
as
previously
mentioned), and then upon committing, we
just
write
the
cached
records
as a
whole into the store and then call flush.
Guozhang
On Tue, May 31, 2022 at 4:08 PM Alexander
Sorokoumov
<asorokou...@confluent.io.invalid>
wrote:
Hey,
Thank you for the wealth of great
suggestions
and
questions!
I
am
going
to
address the feedback in batches and
update
the
proposal
async,
as
it is
probably going to be easier for
everyone.
I
will
also
write
a
separate
message after making updates to the KIP.
@John,
Did you consider instead just adding
the
option
to
the
RocksDB*StoreSupplier classes and the
factories
in
Stores ?
Thank you for suggesting that. I think
that
this
idea
is
better
than
what I
came up with and will update the KIP
with
configuring
transactionality
via
the suppliers and Stores.
what is the advantage over just doing
the
same
thing
with
the
RecordCache
and not introducing the WriteBatch at
all?
Can you point me to RecordCache? I can't
find
it
in
the
project.
The
advantage would be that WriteBatch
guarantees
write
atomicity.
As
far
as
I
understood the way RecordCache works, it
might
leave
the
system
in
an
inconsistent state during crash failure
on
write.
You mentioned that a transactional store
can
help
reduce
duplication in
the
case of ALOS
I will remove claims about ALOS from the
proposal.
Thank
you
for
elaborating!
As a reminder, we have a new IQv2
mechanism
now.
Should
we
propose
any
changes to IQv1 to support this
transactional
mechanism,
versus
just
proposing it for IQv2? Certainly, it
seems
strange
only
to
propose a
change
for IQv1 and not v2.
I will update the proposal with
complementary
API
changes
for
IQv2
What should IQ do if I request to
readCommitted
on a
non-transactional
store?
We can assume that non-transactional
stores
commit
on
write,
so
IQ
works
in
the same way with non-transactional
stores
regardless
of
the
value
of
readCommitted.
@Guozhang,
* If we crash between line 3 and 4, then
at
that
time
the
local
persistent
store image is representing as of
offset
200,
but
upon
recovery
all
changelog records from 100 to
log-end-offset
would
be
considered
as
aborted
and not be replayed and we would
restart
processing
from
position
100.
Restart processing will violate EOS.I'm
not
sure
how
e.g.
RocksDB's
WriteBatchWithIndex would make sure
that
the
step 4
and
step 5
could
be
done atomically here.
Could you please point me to the place
in
the
codebase
where
a
task
flushes
the store before committing the
transaction?
Looking at TaskExecutor (
https://github.com/apache/kafka/blob/4c9eeef5b2dff9a4f0977fbc5ac7eaaf930d0d0e/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java#L144-L167
),
StreamTask#prepareCommit (
https://github.com/apache/kafka/blob/4c9eeef5b2dff9a4f0977fbc5ac7eaaf930d0d0e/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L398
),
and CachedStateStore (
https://github.com/apache/kafka/blob/4c9eeef5b2dff9a4f0977fbc5ac7eaaf930d0d0e/streams/src/main/java/org/apache/kafka/streams/state/internals/CachedStateStore.java#L29-L34
)
we flush the cache, but not the
underlying
state
store.
Explicit
StateStore#flush happens in
AbstractTask#maybeWriteCheckpoint (
https://github.com/apache/kafka/blob/4c9eeef5b2dff9a4f0977fbc5ac7eaaf930d0d0e/streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java#L91-L99
).
Is there something I am missing here?
Today all cached data that have not been
flushed
are
not
committed
for
sure, but even flushed data to the
persistent
underlying
store
may
also
be
uncommitted since flushing can be
triggered
asynchronously
before
the
commit.
Can you please point me to the place in
the
codebase
where
we
trigger
async
flush before the commit? This would
certainly
be a
reason
to
introduce
a
dedicated StateStore#commit method.
Thanks again for the feedback. I am
going
to
update
the
KIP
and
then
respond to the next batch of questions
and
suggestions.
Best,
Alex
On Mon, May 30, 2022 at 5:13 PM Suhas
Satish
<ssat...@confluent.io.invalid
wrote:
Thanks for the KIP proposal Alex.
1. Configuration default
You mention applications using streams
DSL
with
built-in
rocksDB
state
store will get transactional state
stores
by
default
when
EOS
is
enabled,
but the default implementation for apps
using
PAPI
will
fallback
to
non-transactional behavior.
Shouldn't we have the same default
behavior
for
both
types
of
apps -
DSL
and PAPI?
On Mon, May 30, 2022 at 2:11 AM Bruno
Cadonna <
cado...@apache.org
wrote:
Thanks for the PR, Alex!
I am also glad to see this coming.
1. Configuration
I would also prefer to restrict the
configuration
of
transactional
on
the state sore. Ideally, calling
method
transactional()
on
the
state
store would be enough. An option on
the
store
builder
would
make it
possible to turn transactionality on
and
off
(as
John
proposed).
2. Memory usage in RocksDB
This seems to be a major issue. We do
not
have
any
guarantee
that
uncommitted writes fit into memory
and I
guess
we
will
never
have.
What
happens when the uncommitted writes do
not
fit
into
memory?
Does
RocksDB
throw an exception? Can we handle such
an
exception
without
crashing?
Does the RocksDB behavior even need to
be
included
in
this
KIP?
In
the
end it is an implementation detail.
What we should consider - though - is
a
memory
limit
in
some
form.
And
what we do when the memory limit is
exceeded.
3. PoC
I agree with Guozhang that a PoC is a
good
idea
to
better
understand
the
devils in the details.
Best,
Bruno
On 25.05.22 01:52, Guozhang Wang
wrote:
Hello Alex,
Thanks for writing the proposal! Glad
to
see
it
coming. I
think
this
is
the
kind of a KIP that since too many
devils
would
be
buried
in
the
details
and
it's better to start working on a
POC,
either
in
parallel,
or
before
we
resume our discussion, rather than
blocking
any
implementation
until
we
are
satisfied with the proposal.
Just as a concrete example, I
personally
am
still
not
100%
clear
how
the
proposal would work to achieve EOS
with
the
state
stores.
For
example,
the
commit procedure today looks like
this:
0: there's an existing checkpoint
file
indicating
the
changelog
offset
of
the local state store image is 100.
Now a
commit
is
triggered:
1. flush cache (since it contains
partially
processed
records),
make
sure
all records are written to the
producer.
2. flush producer, making sure all
changelog
records
have
now
acked.
//
here we would get the new changelog
position,
say
200
3. flush store, make sure all writes
are
persisted.
4.
producer.sendOffsetsToTransaction();
producer.commitTransaction();
//
we
would make the writes in changelog up
to
offset
200
committed
5. Update the checkpoint file to 200.
The question about atomicity between
those
lines,
for
example:
* If we crash between line 4 and line
5,
the
local
checkpoint
file
would
stay as 100, and upon recovery we
would
replay
the
changelog
from
100
to
200. This is not ideal but does not
violate
EOS,
since
the
changelogs
are
all overwrites anyways.
* If we crash between line 3 and 4,
then
at
that
time
the
local
persistent
store image is representing as of
offset
200,
but
upon
recovery
all
changelog records from 100 to
log-end-offset
would
be
considered
as
aborted
and not be replayed and we would
restart
processing
from
position
100.
Restart processing will violate
EOS.I'm
not
sure
how
e.g.
RocksDB's
WriteBatchWithIndex would make sure
that
the
step 4
and
step 5
could
be
done atomically here.
Originally what I was thinking when
creating
the
JIRA
ticket
is
that
we
need to let the state store to
provide
a
transactional
API
like
"token
commit()" used in step 4) above which
returns a
token,
that
e.g.
in
our
example above indicates offset 200,
and
that
token
would
be
written
as
part
of the records in Kafka transaction
in
step
5).
And
upon
recovery
the
state
store would have another API like
"rollback(token)"
where
the
token
is
read
from the latest committed txn, and be
used
to
rollback
the
store
to
that
committed image. I think your
proposal
is
different,
and
it
seems
like
you're proposing we swap step 3) and
4)
above,
but
the
atomicity
issue
still remains since now you may have
the
store
image
at
100
but
the
changelog is committed at 200. I'd
like
to
learn
more
about
the
details
on how it resolves such issues.
Anyways, that's just an example to
make
the
point
that
there
are
lots
of
implementational details which would
drive
the
public
API
design,
and
we
should probably first do a POC, and
come
back
to
discuss
the
KIP.
Let
me
know what you think?
Guozhang
On Tue, May 24, 2022 at 10:35 AM
Sagar
<
sagarmeansoc...@gmail.com>
wrote:
Hi Alexander,
Thanks for the KIP! This seems like
a
great
proposal.
I
have
the
same
opinion as John on the Configuration
part
though.
I
think
the 2
level
config and its behaviour based on
the
setting/unsetting
of
the
flag
seems
confusing to me as well. Since the
KIP
seems
specifically
centred
around
RocksDB it might be better to add it
at
the
Supplier
level
as
John
suggested.
On similar lines, this config name
=>
*statestore.transactional.mechanism
*may
also need rethinking as the value
assigned
to
it(rocksdb_indexbatch)
implicitly seems to assume that
rocksdb
is
the
only
statestore
that
Kafka
Stream supports while that's not the
case.
Also, regarding the potential memory
pressure
that
can be
introduced
by
WriteBatchIndex, do you think it
might
make
more
sense to
include
some
numbers/benchmarks on how much the
memory
consumption
might
increase?
Lastly, the read_uncommitted flag's
behaviour
on
IQ
may
need
more
elaboration.
These points aside, as I said, this
is a
great
proposal!
Thanks!
Sagar.
On Tue, May 24, 2022 at 10:35 PM
John
Roesler
<
vvcep...@apache.org>
wrote:
Thanks for the KIP, Alex!
I'm really happy to see your
proposal.
This
improvement
fills a
long-standing gap.
I have a few questions:
1. Configuration
The KIP only mentions RocksDB, but
of
course,
Streams
also
ships
with
an
InMemory store, and users also plug
in
their
own
custom
state
stores.
It
is
also common to use multiple types
of
state
stores
in
the
same
application
for different purposes.
Against this backdrop, the choice
to
configure
transactionality
as
a
top-level config, as well as to
configure
the
store
transaction
mechanism
as a top-level config, seems a bit
off.
Did you consider instead just
adding
the
option
to
the
RocksDB*StoreSupplier classes and
the
factories
in
Stores
?
It
seems
like
the desire to enable the feature by
default,
but
with a
feature-flag
to
disable it was a factor here.
However,
as
you
pointed
out,
there
are
some
major considerations that users
should
be
aware
of,
so
opt-in
doesn't
seem
like a bad choice, either. You
could
add
an
Enum
argument
to
those
factories like
`RocksDBTransactionalMechanism.{NONE,
Some points in favor of this
approach:
* Avoid "stores that don't support
transactions
ignore
the
config"
complexity
* Users can choose how to spend
their
memory
budget,
making
some
stores
transactional and others not
* When we add transactional support
to
in-memory
stores,
we
don't
have
to
figure out what to do with the
mechanism
config
(i.e.,
what
do
you
set
the
mechanism to when there are
multiple
kinds
of
transactional
stores
in
the
topology?)
2. caching/flushing/transactions
The coupling between memory usage
and
flushing
that
you
mentioned
is
a
bit
troubling. It also occurs to me
that
there
seems
to
be
some
relationship
with the existing record cache,
which
is
also
an
in-memory
holding
area
for
records that are not yet written to
the
cache
and/or
store
(albeit
with
no
particular semantics). Have you
considered
how
all
these
components
should
relate? For example, should a
"full"
WriteBatch
actually
trigger
a
flush
so
that we don't get OOMEs? If the
proposed
transactional
mechanism
forces
all
uncommitted writes to be buffered
in
memory,
until
a
commit,
then
what
is
the advantage over just doing the
same
thing
with
the
RecordCache
and
not
introducing the WriteBatch at all?
3. ALOS
You mentioned that a transactional
store
can
help
reduce
duplication
in
the case of ALOS. We might want to
be
careful
about
claims
like
that.
Duplication isn't the way that
repeated
processing
manifests in
state
stores. Rather, it is in the form
of
dirty
reads
during
reprocessing.
This
feature may reduce the incidence of
dirty
reads
during
reprocessing,
but
not in a predictable way. During
regular
processing
today,
we
will
send
some records through to the
changelog
in
between
commit
intervals.
Under
ALOS, if any of those dirty writes
gets
committed
to
the
changelog
topic,
then upon failure, we have to roll
the
store
forward
to
them
anyway,
regardless of this new
transactional
mechanism.
That's a
fixable
problem,
by the way, but this KIP doesn't
seem
to
fix
it.
I
wonder
if we
should
make
any claims about the relationship
of
this
feature
to
ALOS
if
the
real-world
behavior is so complex.
4. IQ
As a reminder, we have a new IQv2
mechanism
now.
Should
we
propose
any
changes to IQv1 to support this
transactional
mechanism,
versus
just
proposing it for IQv2? Certainly,
it
seems
strange
only
to
propose
a
change
for IQv1 and not v2.
Regarding your proposal for IQv1,
I'm
unsure
what
the
behavior
should
be
for readCommitted, since the
current
behavior
also
reads
out of
the
RecordCache. I guess if
readCommitted==false,
then
we
will
continue
to
read
from the cache first, then the
Batch,
then
the
store;
and
if
readCommitted==true, we would skip
the
cache
and
the
Batch
and
only
read
from the persistent RocksDB store?
What should IQ do if I request to
readCommitted
on
a
non-transactional
store?
Thanks again for proposing the KIP,
and
my
apologies
for
the
long
reply;
I'm hoping to air all my concerns
in
one
"batch"
to
save
time
for
you.
Thanks,
-John
On Tue, May 24, 2022, at 03:45,
Alexander
Sorokoumov
wrote:
Hi all,
I've written a KIP for making
Kafka
Streams
state
stores
transactional
and
would like to start a discussion:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-844%3A+Transactional+State+Stores
Best,
Alex
--
[image: Confluent] <
https://www.confluent.io
Suhas Satish
Engineering Manager
Follow us: [image: Blog]
<
https://www.confluent.io/blog?utm_source=footer&utm_medium=email&utm_campaign=ch.email-signature_type.community_content.blog
[image:
Twitter] <
https://twitter.com/ConfluentInc
[image:
LinkedIn]
<
https://www.linkedin.com/company/confluent/
[image: Try Confluent Cloud for Free]
<
https://www.confluent.io/get-started?utm_campaign=tm.fm-apac_cd.inbound&utm_source=gmail&utm_medium=organic
--
-- Guozhang
--
-- Guozhang
--
-- Guozhang
--
-- Guozhang
--
-- Guozhang
--
-- Guozhang
