Hi Bruno,
Thanks for reviewing the KIP. It's been a long road, I started working on
this more than a year ago, and most of the time in the last 6 months has
been spent on the "Atomic Checkpointing" stuff that's been benched, so some
of the reasoning behind some of my decisions have been lost, but I'll do my
best to reconstruct them.
1.
IIRC, this was the initial approach I tried. I don't remember the exact
reasons I changed it to use a separate "view" of the StateStore that
encapsulates the transaction, but I believe it had something to do with
concurrent access to the StateStore from Interactive Query threads. Reads
from interactive queries need to be isolated from the currently ongoing
transaction, both for consistency (so interactive queries don't observe
changes that are subsequently rolled-back), but also to prevent Iterators
opened by an interactive query from being closed and invalidated by the
StreamThread when it commits the transaction, which causes your interactive
queries to crash.
Another reason I believe I implemented it this way was a separation of
concerns. Recall that newTransaction() originally created an object of type
Transaction, not StateStore. My intent was to improve the type-safety of
the API, in an effort to ensure Transactions weren't used incorrectly.
Unfortunately, this didn't pan out, but newTransaction() remained.
Finally, this had the added benefit that implementations could easily add
support for transactions *without* re-writing their existing,
non-transactional implementation. I think this can be a benefit both for
implementers of custom StateStores, but also for anyone extending
RocksDbStore, as they can rely on the existing access methods working how
they expect them to.
I'm not too happy with the way the current design has panned out, so I'm
open to ideas on how to improve it. Key to this is finding some way to
ensure that reads from Interactive Query threads are properly isolated from
the transaction, *without* the performance overhead of checking which
thread the method is being called from on every access.
As for replacing flush() with commit() - I saw no reason to add this
complexity to the KIP, unless there was a need to add arguments to the
flush/commit method. This need arises with Atomic Checkpointing, but that
will be implemented separately, in a future KIP. Do you see a need for some
arguments to the flush/commit method that I've missed? Or were you simply
suggesting a rename?
2.
This is simply due to the practical reason that isolationLevel() is really
a proxy for checking if the app is under EOS. The application configuration
is not provided to the constructor of StateStores, but it *is* provided to
init(), via StateStoreContext. For this reason, it seemed somewhat natural
to add it to StateStoreContext. I think this makes sense, since the
IsolationLevel of all StateStores in an application *must* be the same, and
since those stores are all initialized with the same StateStoreContext, it
seems natural for that context to carry the desired IsolationLevel to use.
3.
Using IsolationLevel instead of just passing `boolean eosEnabled`, like
much of the internals was an attempt to logically de-couple the StateStore
API from the internals of Kafka Streams. Technically, StateStores don't
need to know/care what processing mode the KS app is using, all they need
to know is the isolation level expected of them.
Having formal definitions for the expectations of the two required
IsolationLevels allow implementers to implement transactional stores
without having to dig through the internals of Kafka Streams and understand
exactly how they are used. The tight coupling between state stores and
internal behaviour has actually significantly hindered my progress on this
KIP, and encouraged me to avoid increasing this logical coupling as much as
possible.
This also frees implementations to satisfy those requirements in any way
they choose. Transactions might not be the only/available approach to an
implementation, but they might have an alternative way to satisfy the
isolation requirements. I admit that this point is more about semantics,
but "transactional" would need to be formally defined in order for
implementers to provide a valid implementation, and these IsolationLevels
provide that formal definition.
4.
I can remove them. I added them only as I planned to include them in the
org.apache.kafka.streams.state package, as a recommended base
implementation for all StateStores, including those implemented by users. I
had assumed that anything in "public" packages, such as
org.apache.kafka.streams.state, should be included in a KIP. Is that wrong?
5.
RocksDB provides no way to measure the actual size of a
WriteBatch(WithIndex), so we're limited to tracking the sum total of the
size of keys + values that are written to the transaction. This obviously
under-estimates the actual memory usage, because WriteBatch no-doubt
includes some record overheads, and WriteBatchWithIndex has to maintain an
index.
Ideally, we could trivially add a method upstream to WriteBatchInterface
that provides the exact size of the batch, but that would require an
upgrade of RocksDB, which won't happen soon. So for the time being, we're
stuck with an approximation, so I felt that the new method should reflect
that.
Would you prefer the new method name ignores this constraint and that we
simply make the rocks measurement more accurate in the future?
6.
Done
7.
Very good point. The KIP already specifically calls out memory in the
documentation of the config: "Maximum number of memory bytes to be used to
buffer uncommitted state-store records." - did you have something else in
mind?
Should we also make this clearer by renaming the config property itself?
Perhaps to something like statestore.transaction.buffer.max.bytes?
8.
OK, I can remove this. The intent here was to describe how Streams itself
will manage transaction roll-over etc. Presumably that means we also don't
need a description of how Streams will manage the commit of changelog
transactions, state store transactions and checkpointing?
9.
What do you mean by fail-over? Do you mean failing over an Active Task to
an instance already hosting a Standby Task?
Thanks again and sorry for the essay of a response!
Regards,
Nick
On Tue, 20 Jun 2023 at 10:49, Bruno Cadonna <cado...@apache.org> wrote:
> Hi Nick,
>
> Thanks for the updates!
>
> I really appreciate that you simplified the KIP by removing some
> aspects. As I have already told you, I think the removed aspects are
> also good ideas and we can discuss them on follow-up KIPs.
>
> Regarding the current KIP, I have the following feedback.
>
> 1.
> Is there a good reason to add method newTransaction() to the StateStore
> interface? As far as I understand, the idea is that users of a state
> store (transactional or not) call this method at start-up and after each
> commit. Since the call to newTransaction() is done in any case and I
> think it would simplify the caller code if we just start a new
> transaction after a commit in the implementation?
> As far as I understand, you plan to commit the transaction in the
> flush() method. I find the idea to replace flush() with commit()
> presented in KIP-844 an elegant solution.
>
> 2.
> Why is the method to query the isolation level added to the state store
> context?
>
> 3.
> Do we need all the isolation level definitions? I think it is good to
> know the guarantees of the transactionality of the state store.
> However, currently, Streams guarantees that there will only be one
> transaction that writes to the state store. Only the stream thread that
> executes the active task that owns the state store will write to the
> state store. I think it should be enough to know if the state store is
> transactional or not. So my proposal would be to just add a method on
> the state store interface the returns if a state store is transactional
> or not by returning a boolean or an enum.
>
> 4.
> I am wondering why AbstractTransaction and AbstractTransactionalStore
> are part of the KIP. They look like implementation details that should
> not be exposed in the public API.
>
> 5.
> Why does StateStore#approximateNumUncommittedBytes() return an
> approximate number of bytes?
>
> 6.
> RocksDB is just one implementation of the state stores in Streams.
> However, the issues regarding OOM errors might also apply to other
> custom implementations. So in the KIP I would extract that part from
> section "RocksDB Transaction". I would also move section "RocksDB
> Transaction" to the end of section "Proposed Changes" and handle it as
> an example implementation for a state store.
>
> 7.
> Should statestore.uncommitted.max.bytes only limit the uncommitted bytes
> or the uncommitted bytes that reside in memory? In future, other
> transactional state store implementations might implement a buffer for
> uncommitted records that are able to spill records on disk. I think
> statestore.uncommitted.max.bytes needs to limit the uncommitted bytes
> irrespective if they reside in memory or disk. Since Streams will use
> this config to decide if it needs to trigger a commit, state store
> implementations that can spill to disk will never be able to spill to
> disk. You would only need to change the doc of the config, if you agree
> with me.
>
> 8.
> Section "Transaction Management" about the wrappers is rather a
> implementation detail that should not be in the KIP.
>
> 9.
> Could you add a section that describes how failover will work with the
> transactional state stores? I think section "Error handling" is already
> a good start.
>
>
> Best,
> Bruno
>
>
>
>
> On 15.05.23 11:04, Nick Telford wrote:
> > Hi everyone,
> >
> > Quick update: I've added a new section to the KIP: "Offsets for Consumer
> > Rebalances", that outlines my solution to the problem that
> > StreamsPartitionAssignor needs to read StateStore offsets even if they're
> > not currently open.
> >
> > Regards,
> > Nick
> >
> > On Wed, 3 May 2023 at 11:34, Nick Telford <nick.telf...@gmail.com>
> wrote:
> >
> >> Hi Bruno,
> >>
> >> Thanks for reviewing my proposal.
> >>
> >> 1.
> >> The main reason I added it was because it was easy to do. If we see no
> >> value in it, I can remove it.
> >>
> >> 2.
> >> Global StateStores can have multiple partitions in their input topics
> >> (which function as their changelogs), so they would have more than one
> >> partition.
> >>
> >> 3.
> >> That's a good point. At present, the only method it adds is
> >> isolationLevel(), which is likely not necessary outside of StateStores.
> >> It *does* provide slightly different guarantees in the documentation to
> >> several of the methods (hence the overrides). I'm not sure if this is
> >> enough to warrant a new interface though.
> >> I think the question that remains is whether this interface makes it
> >> easier to implement custom transactional StateStores than if we were to
> >> remove it? Probably not.
> >>
> >> 4.
> >> The main motivation for the Atomic Checkpointing is actually
> performance.
> >> My team has been testing out an implementation of this KIP without it,
> and
> >> we had problems with RocksDB doing *much* more compaction, due to the
> >> significantly increased flush rate. It was enough of a problem that (for
> >> the time being), we had to revert back to Kafka Streams proper.
> >> I think the best way to solve this, as you say, is to keep the
> .checkpoint
> >> files *in addition* to the offsets being stored within the store itself.
> >> Essentially, when closing StateStores, we force a memtable flush, then
> >> call getCommittedOffsets and write those out to the .checkpoint file.
> >> That would ensure the metadata is available to the
> >> StreamsPartitionAssignor for all closed stores.
> >> If there's a crash (no clean close), then we won't be able to guarantee
> >> which offsets were flushed to disk by RocksDB, so we'd need to open (
> >> init()), read offsets, and then close() those stores. But since this is
> >> the exception, and will only occur once (provided it doesn't crash every
> >> time!), I think the performance impact here would be acceptable.
> >>
> >> Thanks for the feedback, please let me know if you have any more
> comments
> >> or questions!
> >>
> >> I'm currently working on rebasing against trunk. This involves adding
> >> support for transactionality to VersionedStateStores. I will probably
> need
> >> to revise my implementation for transactional "segmented" stores, both
> to
> >> accommodate VersionedStateStore, and to clean up some other stuff.
> >>
> >> Regards,
> >> Nick
> >>
> >>
> >> On Tue, 2 May 2023 at 13:45, Bruno Cadonna <cado...@apache.org> wrote:
> >>
> >>> Hi Nick,
> >>>
> >>> Thanks for the updates!
> >>>
> >>> I have a couple of questions/comments.
> >>>
> >>> 1.
> >>> Why do you propose a configuration that involves max. bytes and max.
> >>> reords? I think we are mainly concerned about memory consumption
> because
> >>> we want to limit the off-heap memory used. I cannot think of a case
> >>> where one would want to set the max. number of records.
> >>>
> >>>
> >>> 2.
> >>> Why does
> >>>
> >>> default void commit(final Map<TopicPartition, Long>
> changelogOffsets) {
> >>> flush();
> >>> }
> >>>
> >>> take a map of partitions to changelog offsets?
> >>> The mapping between state stores to partitions is a 1:1 relationship.
> >>> Passing in a single changelog offset should suffice.
> >>>
> >>>
> >>> 3.
> >>> Why do we need the Transaction interface? It should be possible to hide
> >>> beginning and committing a transactions withing the state store
> >>> implementation, so that from outside the state store, it does not
> matter
> >>> whether the state store is transactional or not. What would be the
> >>> advantage of using the Transaction interface?
> >>>
> >>>
> >>> 4.
> >>> Regarding checkpointing offsets, I think we should keep the checkpoint
> >>> file in any case for the reason you mentioned about rebalancing. Even
> if
> >>> that would not be an issue, I would propose to move the change to
> offset
> >>> management to a new KIP and to not add more complexity than needed to
> >>> this one. I would not be too concerned about the consistency violation
> >>> you mention. As far as I understand, with transactional state stores
> >>> Streams would write the checkpoint file during every commit even under
> >>> EOS. In the failure case you describe, Streams would restore the state
> >>> stores from the offsets found in the checkpoint file written during the
> >>> penultimate commit instead of during the last commit. Basically,
> Streams
> >>> would overwrite the records written to the state store between the last
> >>> two commits with the same records read from the changelogs. While I
> >>> understand that this is wasteful, it is -- at the same time --
> >>> acceptable and most importantly it does not break EOS.
> >>>
> >>> Best,
> >>> Bruno
> >>>
> >>>
> >>> On 27.04.23 12:34, Nick Telford wrote:
> >>>> Hi everyone,
> >>>>
> >>>> I find myself (again) considering removing the offset management from
> >>>> StateStores, and keeping the old checkpoint file system. The reason is
> >>> that
> >>>> the StreamPartitionAssignor directly reads checkpoint files in order
> to
> >>>> determine which instance has the most up-to-date copy of the local
> >>> state.
> >>>> If we move offsets into the StateStore itself, then we will need to
> >>> open,
> >>>> initialize, read offsets and then close each StateStore (that is not
> >>>> already assigned and open) for which we have *any* local state, on
> every
> >>>> rebalance.
> >>>>
> >>>> Generally, I don't think there are many "orphan" stores like this
> >>> sitting
> >>>> around on most instances, but even a few would introduce additional
> >>> latency
> >>>> to an already somewhat lengthy rebalance procedure.
> >>>>
> >>>> I'm leaning towards Colt's (Slack) suggestion of just keeping things
> in
> >>> the
> >>>> checkpoint file(s) for now, and not worrying about the race. The
> >>> downside
> >>>> is that we wouldn't be able to remove the explicit RocksDB flush
> >>> on-commit,
> >>>> which likely hurts performance.
> >>>>
> >>>> If anyone has any thoughts or ideas on this subject, I would
> appreciate
> >>> it!
> >>>>
> >>>> Regards,
> >>>> Nick
> >>>>
> >>>> On Wed, 19 Apr 2023 at 15:05, Nick Telford <nick.telf...@gmail.com>
> >>> wrote:
> >>>>
> >>>>> Hi Colt,
> >>>>>
> >>>>> The issue is that if there's a crash between 2 and 3, then you still
> >>> end
> >>>>> up with inconsistent data in RocksDB. The only way to guarantee that
> >>> your
> >>>>> checkpoint offsets and locally stored data are consistent with each
> >>> other
> >>>>> are to atomically commit them, which can be achieved by having the
> >>> offsets
> >>>>> stored in RocksDB.
> >>>>>
> >>>>> The offsets column family is likely to be extremely small (one
> >>>>> per-changelog partition + one per Topology input partition for
> regular
> >>>>> stores, one per input partition for global stores). So the overhead
> >>> will be
> >>>>> minimal.
> >>>>>
> >>>>> A major benefit of doing this is that we can remove the explicit
> calls
> >>> to
> >>>>> db.flush(), which forcibly flushes memtables to disk on-commit. It
> >>> turns
> >>>>> out, RocksDB memtable flushes are largely dictated by Kafka Streams
> >>>>> commits, *not* RocksDB configuration, which could be a major source
> of
> >>>>> confusion. Atomic checkpointing makes it safe to remove these
> explicit
> >>>>> flushes, because it no longer matters exactly when RocksDB flushes
> >>> data to
> >>>>> disk; since the data and corresponding checkpoint offsets will always
> >>> be
> >>>>> flushed together, the local store is always in a consistent state,
> and
> >>>>> on-restart, it can always safely resume restoration from the on-disk
> >>>>> offsets, restoring the small amount of data that hadn't been flushed
> >>> when
> >>>>> the app exited/crashed.
> >>>>>
> >>>>> Regards,
> >>>>> Nick
> >>>>>
> >>>>> On Wed, 19 Apr 2023 at 14:35, Colt McNealy <c...@littlehorse.io>
> >>> wrote:
> >>>>>
> >>>>>> Nick,
> >>>>>>
> >>>>>> Thanks for your reply. Ack to A) and B).
> >>>>>>
> >>>>>> For item C), I see what you're referring to. Your proposed solution
> >>> will
> >>>>>> work, so no need to change it. What I was suggesting was that it
> >>> might be
> >>>>>> possible to achieve this with only one column family. So long as:
> >>>>>>
> >>>>>> - No uncommitted records (i.e. not committed to the changelog)
> are
> >>>>>> *committed* to the state store, AND
> >>>>>> - The Checkpoint offset (which refers to the changelog topic)
> is
> >>> less
> >>>>>> than or equal to the last written changelog offset in rocksdb
> >>>>>>
> >>>>>> I don't see the need to do the full restoration from scratch. My
> >>>>>> understanding was that prior to 844/892, full restorations were
> >>> required
> >>>>>> because there could be uncommitted records written to RocksDB;
> >>> however,
> >>>>>> given your use of RocksDB transactions, that can be avoided with the
> >>>>>> pattern of 1) commit Kafka transaction, 2) commit RocksDB
> >>> transaction, 3)
> >>>>>> update offset in checkpoint file.
> >>>>>>
> >>>>>> Anyways, your proposed solution works equivalently and I don't
> believe
> >>>>>> there is much overhead to an additional column family in RocksDB.
> >>> Perhaps
> >>>>>> it may even perform better than making separate writes to the
> >>> checkpoint
> >>>>>> file.
> >>>>>>
> >>>>>> Colt McNealy
> >>>>>> *Founder, LittleHorse.io*
> >>>>>>
> >>>>>>
> >>>>>> On Wed, Apr 19, 2023 at 5:53 AM Nick Telford <
> nick.telf...@gmail.com>
> >>>>>> wrote:
> >>>>>>
> >>>>>>> Hi Colt,
> >>>>>>>
> >>>>>>> A. I've done my best to de-couple the StateStore stuff from the
> rest
> >>> of
> >>>>>> the
> >>>>>>> Streams engine. The fact that there will be only one ongoing
> (write)
> >>>>>>> transaction at a time is not guaranteed by any API, and is just a
> >>>>>>> consequence of the way Streams operates. To that end, I tried to
> >>> ensure
> >>>>>> the
> >>>>>>> documentation and guarantees provided by the new APIs are
> >>> independent of
> >>>>>>> this incidental behaviour. In practice, you're right, this
> >>> essentially
> >>>>>>> refers to "interactive queries", which are technically "read
> >>>>>> transactions",
> >>>>>>> even if they don't actually use the transaction API to isolate
> >>>>>> themselves.
> >>>>>>>
> >>>>>>> B. Yes, although not ideal. This is for backwards compatibility,
> >>>>>> because:
> >>>>>>> 1) Existing custom StateStore implementations will implement
> >>>>>> flush(),
> >>>>>>> and not commit(), but the Streams engine now calls commit(), so
> those
> >>>>>> calls
> >>>>>>> need to be forwarded to flush() for these legacy stores.
> >>>>>>> 2) Existing StateStore *users*, i.e. outside of the Streams
> >>> engine
> >>>>>>> itself, may depend on explicitly calling flush(), so for these
> cases,
> >>>>>>> flush() needs to be redirected to call commit().
> >>>>>>> If anyone has a better way to guarantee compatibility without
> >>>>>> introducing
> >>>>>>> this potential recursion loop, I'm open to changes!
> >>>>>>>
> >>>>>>> C. This is described in the "Atomic Checkpointing" section. Offsets
> >>> are
> >>>>>>> stored in a separate RocksDB column family, which is guaranteed to
> be
> >>>>>>> atomically flushed to disk with all other column families. The
> issue
> >>> of
> >>>>>>> checkpoints being written to disk after commit causing
> inconsistency
> >>> if
> >>>>>> it
> >>>>>>> crashes in between is the reason why, under EOS, checkpoint files
> are
> >>>>>> only
> >>>>>>> written on clean shutdown. This is one of the major causes of "full
> >>>>>>> restorations", so moving the offsets into a place where they can be
> >>>>>>> guaranteed to be atomically written with the data they checkpoint
> >>>>>> allows us
> >>>>>>> to write the checkpoint offsets *on every commit*, not just on
> clean
> >>>>>>> shutdown.
> >>>>>>>
> >>>>>>> Regards,
> >>>>>>> Nick
> >>>>>>>
> >>>>>>> On Tue, 18 Apr 2023 at 15:39, Colt McNealy <c...@littlehorse.io>
> >>> wrote:
> >>>>>>>
> >>>>>>>> Nick,
> >>>>>>>>
> >>>>>>>> Thank you for continuing this work. I have a few minor clarifying
> >>>>>>>> questions.
> >>>>>>>>
> >>>>>>>> A) "Records written to any transaction are visible to all other
> >>>>>>>> transactions immediately." I am confused here—I thought there
> could
> >>>>>> only
> >>>>>>> be
> >>>>>>>> one transaction going on at a time for a given state store given
> the
> >>>>>>>> threading model for processing records on a Task. Do you mean
> >>>>>> Interactive
> >>>>>>>> Queries by "other transactions"? (If so, then everything makes
> >>> sense—I
> >>>>>>>> thought that since IQ were read-only then they didn't count as
> >>>>>>>> transactions).
> >>>>>>>>
> >>>>>>>> B) Is it intentional that the default implementations of the
> flush()
> >>>>>> and
> >>>>>>>> commit() methods in the StateStore class refer to each other in
> some
> >>>>>> sort
> >>>>>>>> of unbounded recursion?
> >>>>>>>>
> >>>>>>>> C) How will the getCommittedOffset() method work? At first I
> thought
> >>>>>> the
> >>>>>>>> way to do it would be using a special key in the RocksDB store to
> >>>>>> store
> >>>>>>> the
> >>>>>>>> offset, and committing that with the transaction. But upon second
> >>>>>>> thought,
> >>>>>>>> since restoration from the changelog is an idempotent procedure, I
> >>>>>> think
> >>>>>>> it
> >>>>>>>> would be fine to 1) commit the RocksDB transaction and then 2)
> write
> >>>>>> the
> >>>>>>>> offset to disk in a checkpoint file. If there is a crash between
> 1)
> >>>>>> and
> >>>>>>> 2),
> >>>>>>>> I think the only downside is now we replay a few more records (at
> a
> >>>>>> cost
> >>>>>>> of
> >>>>>>>> <100ms). Am I missing something there?
> >>>>>>>>
> >>>>>>>> Other than that, everything makes sense to me.
> >>>>>>>>
> >>>>>>>> Cheers,
> >>>>>>>> Colt McNealy
> >>>>>>>> *Founder, LittleHorse.io*
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> On Tue, Apr 18, 2023 at 3:59 AM Nick Telford <
> >>> nick.telf...@gmail.com>
> >>>>>>>> wrote:
> >>>>>>>>
> >>>>>>>>> Hi everyone,
> >>>>>>>>>
> >>>>>>>>> I've updated the KIP to reflect the latest version of the design:
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-892%3A+Transactional+Semantics+for+StateStores
> >>>>>>>>>
> >>>>>>>>> There are several changes in there that reflect feedback from
> this
> >>>>>>>> thread,
> >>>>>>>>> and there's a new section and a bunch of interface changes
> relating
> >>>>>> to
> >>>>>>>>> Atomic Checkpointing, which is the final piece of the puzzle to
> >>>>>> making
> >>>>>>>>> everything robust.
> >>>>>>>>>
> >>>>>>>>> Let me know what you think!
> >>>>>>>>>
> >>>>>>>>> Regards,
> >>>>>>>>> Nick
> >>>>>>>>>
> >>>>>>>>> On Tue, 3 Jan 2023 at 11:33, Nick Telford <
> nick.telf...@gmail.com>
> >>>>>>>> wrote:
> >>>>>>>>>
> >>>>>>>>>> Hi Lucas,
> >>>>>>>>>>
> >>>>>>>>>> Thanks for looking over my KIP.
> >>>>>>>>>>
> >>>>>>>>>> A) The bound is per-instance, not per-Task. This was a typo in
> the
> >>>>>>> KIP
> >>>>>>>>>> that I've now corrected. It was originally per-Task, but I
> >>>>>> changed it
> >>>>>>>> to
> >>>>>>>>>> per-instance for exactly the reason you highlighted.
> >>>>>>>>>> B) It's worth noting that transactionality is only enabled under
> >>>>>> EOS,
> >>>>>>>> and
> >>>>>>>>>> in the default mode of operation (ALOS), there should be no
> >>>>>> change in
> >>>>>>>>>> behavior at all. I think, under EOS, we can mitigate the impact
> on
> >>>>>>>> users
> >>>>>>>>> by
> >>>>>>>>>> sufficiently low default values for the memory bound
> >>>>>> configuration. I
> >>>>>>>>>> understand your hesitation to include a significant change of
> >>>>>>>> behaviour,
> >>>>>>>>>> especially in a minor release, but I suspect that most users
> will
> >>>>>>>> prefer
> >>>>>>>>>> the memory impact (under EOS) to the existing behaviour of
> >>>>>> frequent
> >>>>>>>> state
> >>>>>>>>>> restorations! If this is a problem, the changes can wait until
> the
> >>>>>>> next
> >>>>>>>>>> major release. I'll be running a patched version of streams in
> >>>>>>>> production
> >>>>>>>>>> with these changes as soon as they're ready, so it won't disrupt
> >>>>>> me
> >>>>>>> :-D
> >>>>>>>>>> C) The main purpose of this sentence was just to note that some
> >>>>>>> changes
> >>>>>>>>>> will need to be made to the way Segments are handled in order to
> >>>>>>> ensure
> >>>>>>>>>> they also benefit from transactions. At the time I wrote it, I
> >>>>>> hadn't
> >>>>>>>>>> figured out the specific changes necessary, so it was
> deliberately
> >>>>>>>> vague.
> >>>>>>>>>> This is the one outstanding problem I'm currently working on,
> and
> >>>>>>> I'll
> >>>>>>>>>> update this section with more detail once I have figured out the
> >>>>>>> exact
> >>>>>>>>>> changes required.
> >>>>>>>>>> D) newTransaction() provides the necessary isolation guarantees.
> >>>>>>> While
> >>>>>>>>>> the RocksDB implementation of transactions doesn't technically
> >>>>>> *need*
> >>>>>>>>>> read-only users to call newTransaction(), other implementations
> >>>>>>> (e.g. a
> >>>>>>>>>> hypothetical PostgresStore) may require it. Calling
> >>>>>> newTransaction()
> >>>>>>>> when
> >>>>>>>>>> no transaction is necessary is essentially free, as it will just
> >>>>>>> return
> >>>>>>>>>> this.
> >>>>>>>>>>
> >>>>>>>>>> I didn't do any profiling of the KIP-844 PoC, but I think it
> >>>>>> should
> >>>>>>> be
> >>>>>>>>>> fairly obvious where the performance problems stem from: writes
> >>>>>> under
> >>>>>>>>>> KIP-844 require 3 extra memory-copies: 1 to encode it with the
> >>>>>>>>>> tombstone/record flag, 1 to decode it from the tombstone/record
> >>>>>> flag,
> >>>>>>>>> and 1
> >>>>>>>>>> to copy the record from the "temporary" store to the "main"
> store,
> >>>>>>> when
> >>>>>>>>> the
> >>>>>>>>>> transaction commits. The different approach taken by KIP-869
> >>>>>> should
> >>>>>>>>> perform
> >>>>>>>>>> much better, as it avoids all these copies, and may actually
> >>>>>> perform
> >>>>>>>>>> slightly better than trunk, due to batched writes in RocksDB
> >>>>>>> performing
> >>>>>>>>>> better than non-batched writes.[1]
> >>>>>>>>>>
> >>>>>>>>>> Regards,
> >>>>>>>>>> Nick
> >>>>>>>>>>
> >>>>>>>>>> 1:
> >>>>>>>>>
> >>>>>>>
> >>>
> https://github.com/adamretter/rocksjava-write-methods-benchmark#results
> >>>>>>>>>>
> >>>>>>>>>> On Mon, 2 Jan 2023 at 16:18, Lucas Brutschy <
> >>>>>> lbruts...@confluent.io
> >>>>>>>>> .invalid>
> >>>>>>>>>> wrote:
> >>>>>>>>>>
> >>>>>>>>>>> Hi Nick,
> >>>>>>>>>>>
> >>>>>>>>>>> I'm just starting to read up on the whole discussion about
> >>>>>> KIP-892
> >>>>>>> and
> >>>>>>>>>>> KIP-844. Thanks a lot for your work on this, I do think
> >>>>>>>>>>> `WriteBatchWithIndex` may be the way to go here. I do have some
> >>>>>>>>>>> questions about the latest draft.
> >>>>>>>>>>>
> >>>>>>>>>>> A) If I understand correctly, you propose to put a bound on
> the
> >>>>>>>>>>> (native) memory consumed by each task. However, I wonder if
> this
> >>>>>> is
> >>>>>>>>>>> sufficient if we have temporary imbalances in the cluster. For
> >>>>>>>>>>> example, depending on the timing of rebalances during a cluster
> >>>>>>>>>>> restart, it could happen that a single streams node is
> assigned a
> >>>>>>> lot
> >>>>>>>>>>> more tasks than expected. With your proposed change, this would
> >>>>>> mean
> >>>>>>>>>>> that the memory required by this one node could be a multiple
> of
> >>>>>>> what
> >>>>>>>>>>> is required during normal operation. I wonder if it wouldn't be
> >>>>>>> safer
> >>>>>>>>>>> to put a global bound on the memory use, across all tasks.
> >>>>>>>>>>> B) Generally, the memory concerns still give me the feeling
> >>> that
> >>>>>>> this
> >>>>>>>>>>> should not be enabled by default for all users in a minor
> >>>>>> release.
> >>>>>>>>>>> C) In section "Transaction Management": the sentence "A
> similar
> >>>>>>>>>>> analogue will be created to automatically manage `Segment`
> >>>>>>>>>>> transactions.". Maybe this is just me lacking some background,
> >>>>>> but I
> >>>>>>>>>>> do not understand this, it would be great if you could clarify
> >>>>>> what
> >>>>>>>>>>> you mean here.
> >>>>>>>>>>> D) Could you please clarify why IQ has to call
> >>> newTransaction(),
> >>>>>>> when
> >>>>>>>>>>> it's read-only.
> >>>>>>>>>>>
> >>>>>>>>>>> And one last thing not strictly related to your KIP: if there
> is
> >>>>>> an
> >>>>>>>>>>> easy way for you to find out why the KIP-844 PoC is 20x slower
> >>>>>> (e.g.
> >>>>>>>>>>> by providing a flame graph), that would be quite interesting.
> >>>>>>>>>>>
> >>>>>>>>>>> Cheers,
> >>>>>>>>>>> Lucas
> >>>>>>>>>>>
> >>>>>>>>>>> On Thu, Dec 22, 2022 at 8:30 PM Nick Telford <
> >>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>> wrote:
> >>>>>>>>>>>>
> >>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>
> >>>>>>>>>>>> I've updated the KIP with a more detailed design, which
> >>>>>> reflects
> >>>>>>> the
> >>>>>>>>>>>> implementation I've been working on:
> >>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-892%3A+Transactional+Semantics+for+StateStores
> >>>>>>>>>>>>
> >>>>>>>>>>>> This new design should address the outstanding points already
> >>>>>> made
> >>>>>>>> in
> >>>>>>>>>>> the
> >>>>>>>>>>>> thread.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Please let me know if there are areas that are unclear or need
> >>>>>>> more
> >>>>>>>>>>>> clarification.
> >>>>>>>>>>>>
> >>>>>>>>>>>> I have a (nearly) working implementation. I'm confident that
> >>>>>> the
> >>>>>>>>>>> remaining
> >>>>>>>>>>>> work (making Segments behave) will not impact the documented
> >>>>>>> design.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Regards,
> >>>>>>>>>>>>
> >>>>>>>>>>>> Nick
> >>>>>>>>>>>>
> >>>>>>>>>>>> On Tue, 6 Dec 2022 at 19:24, Colt McNealy <
> c...@littlehorse.io
> >>>>>>>
> >>>>>>>>> wrote:
> >>>>>>>>>>>>
> >>>>>>>>>>>>> Nick,
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Thank you for the reply; that makes sense. I was hoping that,
> >>>>>>>> since
> >>>>>>>>>>> reading
> >>>>>>>>>>>>> uncommitted records from IQ in EOS isn't part of the
> >>>>>> documented
> >>>>>>>> API,
> >>>>>>>>>>> maybe
> >>>>>>>>>>>>> you *wouldn't* have to wait for the next major release to
> >>>>>> make
> >>>>>>>> that
> >>>>>>>>>>> change;
> >>>>>>>>>>>>> but given that it would be considered a major change, I like
> >>>>>>> your
> >>>>>>>>>>> approach
> >>>>>>>>>>>>> the best.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Wishing you a speedy recovery and happy coding!
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Thanks,
> >>>>>>>>>>>>> Colt McNealy
> >>>>>>>>>>>>> *Founder, LittleHorse.io*
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> On Tue, Dec 6, 2022 at 10:30 AM Nick Telford <
> >>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>> Hi Colt,
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> 10: Yes, I agree it's not ideal. I originally intended to
> >>>>>> try
> >>>>>>> to
> >>>>>>>>>>> keep the
> >>>>>>>>>>>>>> behaviour unchanged as much as possible, otherwise we'd
> >>>>>> have
> >>>>>>> to
> >>>>>>>>>>> wait for
> >>>>>>>>>>>>> a
> >>>>>>>>>>>>>> major version release to land these changes.
> >>>>>>>>>>>>>> 20: Good point, ALOS doesn't need the same level of
> >>>>>> guarantee,
> >>>>>>>> and
> >>>>>>>>>>> the
> >>>>>>>>>>>>>> typically longer commit intervals would be problematic when
> >>>>>>>>> reading
> >>>>>>>>>>> only
> >>>>>>>>>>>>>> "committed" records.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> I've been away for 5 days recovering from minor surgery,
> >>>>>> but I
> >>>>>>>>>>> spent a
> >>>>>>>>>>>>>> considerable amount of that time working through ideas for
> >>>>>>>>> possible
> >>>>>>>>>>>>>> solutions in my head. I think your suggestion of keeping
> >>>>>> ALOS
> >>>>>>>>>>> as-is, but
> >>>>>>>>>>>>>> buffering writes for EOS is the right path forwards,
> >>>>>> although
> >>>>>>> I
> >>>>>>>>>>> have a
> >>>>>>>>>>>>>> solution that both expands on this, and provides for some
> >>>>>> more
> >>>>>>>>>>> formal
> >>>>>>>>>>>>>> guarantees.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Essentially, adding support to KeyValueStores for
> >>>>>>>> "Transactions",
> >>>>>>>>>>> with
> >>>>>>>>>>>>>> clearly defined IsolationLevels. Using "Read Committed"
> >>>>>> when
> >>>>>>>> under
> >>>>>>>>>>> EOS,
> >>>>>>>>>>>>> and
> >>>>>>>>>>>>>> "Read Uncommitted" under ALOS.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> The nice thing about this approach is that it gives us much
> >>>>>>> more
> >>>>>>>>>>> clearly
> >>>>>>>>>>>>>> defined isolation behaviour that can be properly
> >>>>>> documented to
> >>>>>>>>>>> ensure
> >>>>>>>>>>>>> users
> >>>>>>>>>>>>>> know what to expect.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> I'm still working out the kinks in the design, and will
> >>>>>> update
> >>>>>>>> the
> >>>>>>>>>>> KIP
> >>>>>>>>>>>>> when
> >>>>>>>>>>>>>> I have something. The main struggle is trying to implement
> >>>>>>> this
> >>>>>>>>>>> without
> >>>>>>>>>>>>>> making any major changes to the existing interfaces or
> >>>>>>> breaking
> >>>>>>>>>>> existing
> >>>>>>>>>>>>>> implementations, because currently everything expects to
> >>>>>>> operate
> >>>>>>>>>>> directly
> >>>>>>>>>>>>>> on a StateStore, and not a Transaction of that store. I
> >>>>>> think
> >>>>>>>> I'm
> >>>>>>>>>>> getting
> >>>>>>>>>>>>>> close, although sadly I won't be able to progress much
> >>>>>> until
> >>>>>>>> next
> >>>>>>>>>>> week
> >>>>>>>>>>>>> due
> >>>>>>>>>>>>>> to some work commitments.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> On Thu, 1 Dec 2022 at 00:01, Colt McNealy <
> >>>>>>> c...@littlehorse.io>
> >>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Nick,
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Thank you for the explanation, and also for the updated
> >>>>>>> KIP. I
> >>>>>>>>> am
> >>>>>>>>>>> quite
> >>>>>>>>>>>>>>> eager for this improvement to be released as it would
> >>>>>>> greatly
> >>>>>>>>>>> reduce
> >>>>>>>>>>>>> the
> >>>>>>>>>>>>>>> operational difficulties of EOS streams apps.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Two questions:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> 10)
> >>>>>>>>>>>>>>>> When reading records, we will use the
> >>>>>>>>>>>>>>> WriteBatchWithIndex#getFromBatchAndDB
> >>>>>>>>>>>>>>> and WriteBatchWithIndex#newIteratorWithBase utilities in
> >>>>>>>> order
> >>>>>>>>> to
> >>>>>>>>>>>>> ensure
> >>>>>>>>>>>>>>> that uncommitted writes are available to query.
> >>>>>>>>>>>>>>> Why do extra work to enable the reading of uncommitted
> >>>>>>> writes
> >>>>>>>>>>> during
> >>>>>>>>>>>>> IQ?
> >>>>>>>>>>>>>>> Code complexity aside, reading uncommitted writes is, in
> >>>>>> my
> >>>>>>>>>>> opinion, a
> >>>>>>>>>>>>>>> minor flaw in EOS IQ; it would be very nice to have the
> >>>>>>>>> guarantee
> >>>>>>>>>>> that,
> >>>>>>>>>>>>>>> with EOS, IQ only reads committed records. In order to
> >>>>>> avoid
> >>>>>>>>> dirty
> >>>>>>>>>>>>> reads,
> >>>>>>>>>>>>>>> one currently must query a standby replica (but this
> >>>>>> still
> >>>>>>>>> doesn't
> >>>>>>>>>>>>> fully
> >>>>>>>>>>>>>>> guarantee monotonic reads).
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> 20) Is it also necessary to enable this optimization on
> >>>>>> ALOS
> >>>>>>>>>>> stores?
> >>>>>>>>>>>>> The
> >>>>>>>>>>>>>>> motivation of KIP-844 was mainly to reduce the need to
> >>>>>>> restore
> >>>>>>>>>>> state
> >>>>>>>>>>>>> from
> >>>>>>>>>>>>>>> scratch on unclean EOS shutdowns; with ALOS it was
> >>>>>>> acceptable
> >>>>>>>> to
> >>>>>>>>>>> accept
> >>>>>>>>>>>>>>> that there may have been uncommitted writes on disk. On a
> >>>>>>> side
> >>>>>>>>>>> note, if
> >>>>>>>>>>>>>> you
> >>>>>>>>>>>>>>> enable this type of store on ALOS processors, the
> >>>>>> community
> >>>>>>>>> would
> >>>>>>>>>>>>>>> definitely want to enable queries on dirty reads;
> >>>>>> otherwise
> >>>>>>>>> users
> >>>>>>>>>>> would
> >>>>>>>>>>>>>>> have to wait 30 seconds (default) to see an update.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Thank you for doing this fantastic work!
> >>>>>>>>>>>>>>> Colt McNealy
> >>>>>>>>>>>>>>> *Founder, LittleHorse.io*
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> On Wed, Nov 30, 2022 at 10:44 AM Nick Telford <
> >>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> I've drastically reduced the scope of this KIP to no
> >>>>>>> longer
> >>>>>>>>>>> include
> >>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>> StateStore management of checkpointing. This can be
> >>>>>> added
> >>>>>>>> as a
> >>>>>>>>>>> KIP
> >>>>>>>>>>>>>> later
> >>>>>>>>>>>>>>> on
> >>>>>>>>>>>>>>>> to further optimize the consistency and performance of
> >>>>>>> state
> >>>>>>>>>>> stores.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> I've also added a section discussing some of the
> >>>>>> concerns
> >>>>>>>>> around
> >>>>>>>>>>>>>>>> concurrency, especially in the presence of Iterators.
> >>>>>> I'm
> >>>>>>>>>>> thinking of
> >>>>>>>>>>>>>>>> wrapping WriteBatchWithIndex with a reference-counting
> >>>>>>>>>>> copy-on-write
> >>>>>>>>>>>>>>>> implementation (that only makes a copy if there's an
> >>>>>>> active
> >>>>>>>>>>>>> iterator),
> >>>>>>>>>>>>>>> but
> >>>>>>>>>>>>>>>> I'm open to suggestions.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> On Mon, 28 Nov 2022 at 16:36, Nick Telford <
> >>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Hi Colt,
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> I didn't do any profiling, but the 844
> >>>>>> implementation:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> - Writes uncommitted records to a temporary
> >>>>>> RocksDB
> >>>>>>>>>>> instance
> >>>>>>>>>>>>>>>>> - Since tombstones need to be flagged, all
> >>>>>> record
> >>>>>>>>>>> values are
> >>>>>>>>>>>>>>>>> prefixed with a value/tombstone marker. This
> >>>>>>>>>>> necessitates a
> >>>>>>>>>>>>>>> memory
> >>>>>>>>>>>>>>>> copy.
> >>>>>>>>>>>>>>>>> - On-commit, iterates all records in this
> >>>>>> temporary
> >>>>>>>>>>> instance and
> >>>>>>>>>>>>>>>>> writes them to the main RocksDB store.
> >>>>>>>>>>>>>>>>> - While iterating, the value/tombstone marker
> >>>>>> needs
> >>>>>>> to
> >>>>>>>> be
> >>>>>>>>>>> parsed
> >>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>> the real value extracted. This necessitates
> >>>>>> another
> >>>>>>>>> memory
> >>>>>>>>>>> copy.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> My guess is that the cost of iterating the temporary
> >>>>>>>> RocksDB
> >>>>>>>>>>> store
> >>>>>>>>>>>>> is
> >>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>> major factor, with the 2 extra memory copies
> >>>>>> per-Record
> >>>>>>>>>>>>> contributing
> >>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>> significant amount too.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> On Mon, 28 Nov 2022 at 16:12, Colt McNealy <
> >>>>>>>>>>> c...@littlehorse.io>
> >>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> Hi all,
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> Out of curiosity, why does the performance of the
> >>>>>> store
> >>>>>>>>>>> degrade so
> >>>>>>>>>>>>>>>>>> significantly with the 844 implementation? I
> >>>>>> wouldn't
> >>>>>>> be
> >>>>>>>>> too
> >>>>>>>>>>>>>> surprised
> >>>>>>>>>>>>>>>> by
> >>>>>>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>> 50-60% drop (caused by each record being written
> >>>>>>> twice),
> >>>>>>>>> but
> >>>>>>>>>>> 96%
> >>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>> extreme.
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> The only thing I can think of which could create
> >>>>>> such a
> >>>>>>>>>>> bottleneck
> >>>>>>>>>>>>>>> would
> >>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>> that perhaps the 844 implementation deserializes and
> >>>>>>> then
> >>>>>>>>>>>>>>> re-serializes
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>> store values when copying from the uncommitted to
> >>>>>>>> committed
> >>>>>>>>>>> store,
> >>>>>>>>>>>>>>> but I
> >>>>>>>>>>>>>>>>>> wasn't able to figure that out when I scanned the
> >>>>>> PR.
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> Colt McNealy
> >>>>>>>>>>>>>>>>>> *Founder, LittleHorse.io*
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> On Mon, Nov 28, 2022 at 7:56 AM Nick Telford <
> >>>>>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> I've updated the KIP to resolve all the points
> >>>>>> that
> >>>>>>>> have
> >>>>>>>>>>> been
> >>>>>>>>>>>>>> raised
> >>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>> far, with one exception: the ALOS default commit
> >>>>>>>> interval
> >>>>>>>>>>> of 5
> >>>>>>>>>>>>>>> minutes
> >>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>> likely to cause WriteBatchWithIndex memory to grow
> >>>>>>> too
> >>>>>>>>>>> large.
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> There's a couple of different things I can think
> >>>>>> of
> >>>>>>> to
> >>>>>>>>>>> solve
> >>>>>>>>>>>>> this:
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> - We already have a memory/record limit in the
> >>>>>> KIP
> >>>>>>>> to
> >>>>>>>>>>> prevent
> >>>>>>>>>>>>>> OOM
> >>>>>>>>>>>>>>>>>>> errors. Should we choose a default value for
> >>>>>>> these?
> >>>>>>>> My
> >>>>>>>>>>>>> concern
> >>>>>>>>>>>>>>> here
> >>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>> anything we choose might seem rather
> >>>>>> arbitrary. We
> >>>>>>>>> could
> >>>>>>>>>>>>> change
> >>>>>>>>>>>>>>>>>>> its behaviour such that under ALOS, it only
> >>>>>>> triggers
> >>>>>>>>> the
> >>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>> StateStore, but under EOS, it triggers a
> >>>>>> commit of
> >>>>>>>> the
> >>>>>>>>>>> Kafka
> >>>>>>>>>>>>>>>>>>> transaction.
> >>>>>>>>>>>>>>>>>>> - We could introduce a separate `
> >>>>>>>>> checkpoint.interval.ms`
> >>>>>>>>>>> to
> >>>>>>>>>>>>>>> allow
> >>>>>>>>>>>>>>>>>> ALOS
> >>>>>>>>>>>>>>>>>>> to commit the StateStores more frequently than
> >>>>>> the
> >>>>>>>>>>> general
> >>>>>>>>>>>>>>>>>>> commit.interval.ms? My concern here is that
> >>>>>> the
> >>>>>>>>>>> semantics of
> >>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>> config
> >>>>>>>>>>>>>>>>>>> would depend on the processing.mode; under
> >>>>>> ALOS it
> >>>>>>>>> would
> >>>>>>>>>>>>> allow
> >>>>>>>>>>>>>>> more
> >>>>>>>>>>>>>>>>>>> frequently committing stores, whereas under
> >>>>>> EOS it
> >>>>>>>>>>> couldn't.
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> Any better ideas?
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> On Wed, 23 Nov 2022 at 16:25, Nick Telford <
> >>>>>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> Hi Alex,
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> Thanks for the feedback.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> I've updated the discussion of OOM issues by
> >>>>>>>> describing
> >>>>>>>>>>> how
> >>>>>>>>>>>>>> we'll
> >>>>>>>>>>>>>>>>>> handle
> >>>>>>>>>>>>>>>>>>>> it. Here's the new text:
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> To mitigate this, we will automatically force a
> >>>>>>> Task
> >>>>>>>>>>> commit if
> >>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>> total
> >>>>>>>>>>>>>>>>>>>>> uncommitted records returned by
> >>>>>>>>>>>>>>>>>>>>> StateStore#approximateNumUncommittedEntries()
> >>>>>>>>> exceeds a
> >>>>>>>>>>>>>>> threshold,
> >>>>>>>>>>>>>>>>>>>>> configured by
> >>>>>>>> max.uncommitted.state.entries.per.task;
> >>>>>>>>>>> or the
> >>>>>>>>>>>>>>> total
> >>>>>>>>>>>>>>>>>>>>> memory used for buffering uncommitted records
> >>>>>>>> returned
> >>>>>>>>>>> by
> >>>>>>>>>>>>>>>>>>>>> StateStore#approximateNumUncommittedBytes()
> >>>>>>> exceeds
> >>>>>>>>> the
> >>>>>>>>>>>>>> threshold
> >>>>>>>>>>>>>>>>>>>>> configured by
> >>>>>>> max.uncommitted.state.bytes.per.task.
> >>>>>>>>>>> This will
> >>>>>>>>>>>>>>>> roughly
> >>>>>>>>>>>>>>>>>>>>> bound the memory required per-Task for
> >>>>>> buffering
> >>>>>>>>>>> uncommitted
> >>>>>>>>>>>>>>>> records,
> >>>>>>>>>>>>>>>>>>>>> irrespective of the commit.interval.ms, and
> >>>>>> will
> >>>>>>>>>>> effectively
> >>>>>>>>>>>>>>> bound
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> number of records that will need to be
> >>>>>> restored in
> >>>>>>>> the
> >>>>>>>>>>> event
> >>>>>>>>>>>>>> of a
> >>>>>>>>>>>>>>>>>>> failure.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> These limits will be checked in
> >>>>>> StreamTask#process
> >>>>>>>> and
> >>>>>>>>> a
> >>>>>>>>>>>>>> premature
> >>>>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>> will be requested via Task#requestCommit().
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> Note that these new methods provide default
> >>>>>>>>>>> implementations
> >>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>> ensure
> >>>>>>>>>>>>>>>>>>>>> existing custom stores and non-transactional
> >>>>>>> stores
> >>>>>>>>>>> (e.g.
> >>>>>>>>>>>>>>>>>>>>> InMemoryKeyValueStore) do not force any early
> >>>>>>>> commits.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> I've chosen to have the StateStore expose
> >>>>>>>>> approximations
> >>>>>>>>>>> of
> >>>>>>>>>>>>> its
> >>>>>>>>>>>>>>>> buffer
> >>>>>>>>>>>>>>>>>>>> size/count instead of opaquely requesting a
> >>>>>> commit
> >>>>>>> in
> >>>>>>>>>>> order to
> >>>>>>>>>>>>>>>>>> delegate
> >>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>> decision making to the Task itself. This enables
> >>>>>>>> Tasks
> >>>>>>>>>>> to look
> >>>>>>>>>>>>>> at
> >>>>>>>>>>>>>>>>>> *all*
> >>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>> their StateStores, and determine whether an
> >>>>>> early
> >>>>>>>>> commit
> >>>>>>>>>>> is
> >>>>>>>>>>>>>>>> necessary.
> >>>>>>>>>>>>>>>>>>>> Notably, it enables pre-Task thresholds,
> >>>>>> instead of
> >>>>>>>>>>> per-Store,
> >>>>>>>>>>>>>>> which
> >>>>>>>>>>>>>>>>>>>> prevents Tasks with many StateStores from using
> >>>>>>> much
> >>>>>>>>> more
> >>>>>>>>>>>>> memory
> >>>>>>>>>>>>>>>> than
> >>>>>>>>>>>>>>>>>>> Tasks
> >>>>>>>>>>>>>>>>>>>> with one StateStore. This makes sense, since
> >>>>>>> commits
> >>>>>>>>> are
> >>>>>>>>>>> done
> >>>>>>>>>>>>>>>> by-Task,
> >>>>>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>> by-Store.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> Prizes* for anyone who can come up with a better
> >>>>>>> name
> >>>>>>>>>>> for the
> >>>>>>>>>>>>>> new
> >>>>>>>>>>>>>>>>>> config
> >>>>>>>>>>>>>>>>>>>> properties!
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> Thanks for pointing out the potential
> >>>>>> performance
> >>>>>>>>> issues
> >>>>>>>>>>> of
> >>>>>>>>>>>>>> WBWI.
> >>>>>>>>>>>>>>>> From
> >>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>> benchmarks that user posted[1], it looks like
> >>>>>> WBWI
> >>>>>>>>> still
> >>>>>>>>>>>>>> performs
> >>>>>>>>>>>>>>>>>>>> considerably better than individual puts, which
> >>>>>> is
> >>>>>>>> the
> >>>>>>>>>>>>> existing
> >>>>>>>>>>>>>>>>>> design,
> >>>>>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>>> I'd actually expect a performance boost from
> >>>>>> WBWI,
> >>>>>>>> just
> >>>>>>>>>>> not as
> >>>>>>>>>>>>>>> great
> >>>>>>>>>>>>>>>>>> as
> >>>>>>>>>>>>>>>>>>>> we'd get from a plain WriteBatch. This does
> >>>>>> suggest
> >>>>>>>>> that
> >>>>>>>>>>> a
> >>>>>>>>>>>>> good
> >>>>>>>>>>>>>>>>>>>> optimization would be to use a regular
> >>>>>> WriteBatch
> >>>>>>> for
> >>>>>>>>>>>>>> restoration
> >>>>>>>>>>>>>>>> (in
> >>>>>>>>>>>>>>>>>>>> RocksDBStore#restoreBatch), since we know that
> >>>>>>> those
> >>>>>>>>>>> records
> >>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>> never
> >>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>> queried before they're committed.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> 1:
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>
> >>>>>>
> >>>
> https://github.com/adamretter/rocksjava-write-methods-benchmark#results
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> * Just kidding, no prizes, sadly.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> On Wed, 23 Nov 2022 at 12:28, Alexander
> >>>>>> Sorokoumov
> >>>>>>>>>>>>>>>>>>>> <asorokou...@confluent.io.invalid> wrote:
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> Hey Nick,
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> Thank you for the KIP! With such a significant
> >>>>>>>>>>> performance
> >>>>>>>>>>>>>>>>>> degradation
> >>>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>> the secondary store approach, we should
> >>>>>> definitely
> >>>>>>>>>>> consider
> >>>>>>>>>>>>>>>>>>>>> WriteBatchWithIndex. I also like encapsulating
> >>>>>>>>>>> checkpointing
> >>>>>>>>>>>>>>> inside
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> default state store implementation to improve
> >>>>>>>>>>> performance.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> +1 to John's comment to keep the current
> >>>>>>>> checkpointing
> >>>>>>>>>>> as a
> >>>>>>>>>>>>>>>> fallback
> >>>>>>>>>>>>>>>>>>>>> mechanism. We want to keep existing users'
> >>>>>>> workflows
> >>>>>>>>>>> intact
> >>>>>>>>>>>>> if
> >>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>> can. A
> >>>>>>>>>>>>>>>>>>>>> non-intrusive way would be to add a separate
> >>>>>>>>> StateStore
> >>>>>>>>>>>>> method,
> >>>>>>>>>>>>>>>> say,
> >>>>>>>>>>>>>>>>>>>>> StateStore#managesCheckpointing(), that
> >>>>>> controls
> >>>>>>>>>>> whether the
> >>>>>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>> implementation owns checkpointing.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> I think that a solution to the transactional
> >>>>>>> writes
> >>>>>>>>>>> should
> >>>>>>>>>>>>>>> address
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> OOMEs. One possible way to address that is to
> >>>>>> wire
> >>>>>>>>>>>>> StateStore's
> >>>>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>> request by adding, say, StateStore#commitNeeded
> >>>>>>> that
> >>>>>>>>> is
> >>>>>>>>>>>>> checked
> >>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>> StreamTask#commitNeeded via the corresponding
> >>>>>>>>>>>>>>>> ProcessorStateManager.
> >>>>>>>>>>>>>>>>>>> With
> >>>>>>>>>>>>>>>>>>>>> that change, RocksDBStore will have to track
> >>>>>> the
> >>>>>>>>> current
> >>>>>>>>>>>>>>>> transaction
> >>>>>>>>>>>>>>>>>>> size
> >>>>>>>>>>>>>>>>>>>>> and request a commit when the size goes over a
> >>>>>>>>>>> (configurable)
> >>>>>>>>>>>>>>>>>> threshold.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> AFAIU WriteBatchWithIndex might perform
> >>>>>>>> significantly
> >>>>>>>>>>> slower
> >>>>>>>>>>>>>> than
> >>>>>>>>>>>>>>>>>>> non-txn
> >>>>>>>>>>>>>>>>>>>>> puts as the batch size grows [1]. We should
> >>>>>> have a
> >>>>>>>>>>>>>> configuration
> >>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>> fall
> >>>>>>>>>>>>>>>>>>>>> back to the current behavior (and/or disable
> >>>>>> txn
> >>>>>>>>> stores
> >>>>>>>>>>> for
> >>>>>>>>>>>>>> ALOS)
> >>>>>>>>>>>>>>>>>> unless
> >>>>>>>>>>>>>>>>>>>>> the benchmarks show negligible overhead for
> >>>>>> longer
> >>>>>>>>>>> commits /
> >>>>>>>>>>>>>>>>>>> large-enough
> >>>>>>>>>>>>>>>>>>>>> batch sizes.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> If you prefer to keep the KIP smaller, I would
> >>>>>>>> rather
> >>>>>>>>>>> cut out
> >>>>>>>>>>>>>>>>>>>>> state-store-managed checkpointing rather than
> >>>>>>> proper
> >>>>>>>>>>> OOMe
> >>>>>>>>>>>>>>> handling
> >>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>> being able to switch to non-txn behavior. The
> >>>>>>>>>>> checkpointing
> >>>>>>>>>>>>> is
> >>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>> necessary to solve the recovery-under-EOS
> >>>>>> problem.
> >>>>>>>> On
> >>>>>>>>>>> the
> >>>>>>>>>>>>> other
> >>>>>>>>>>>>>>>> hand,
> >>>>>>>>>>>>>>>>>>> once
> >>>>>>>>>>>>>>>>>>>>> WriteBatchWithIndex is in, it will be much
> >>>>>> easier
> >>>>>>> to
> >>>>>>>>> add
> >>>>>>>>>>>>>>>>>>>>> state-store-managed checkpointing.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> If you share the current implementation, I am
> >>>>>>> happy
> >>>>>>>> to
> >>>>>>>>>>> help
> >>>>>>>>>>>>> you
> >>>>>>>>>>>>>>>>>> address
> >>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> OOMe and configuration parts as well as review
> >>>>>> and
> >>>>>>>>> test
> >>>>>>>>>>> the
> >>>>>>>>>>>>>>> patch.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>>>>>> Alex
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> 1.
> >>>>>> https://github.com/facebook/rocksdb/issues/608
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> On Tue, Nov 22, 2022 at 6:31 PM Nick Telford <
> >>>>>>>>>>>>>>>> nick.telf...@gmail.com
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> Hi John,
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> Thanks for the review and feedback!
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 1. Custom Stores: I've been mulling over this
> >>>>>>>>> problem
> >>>>>>>>>>>>> myself.
> >>>>>>>>>>>>>>> As
> >>>>>>>>>>>>>>>> it
> >>>>>>>>>>>>>>>>>>>>> stands,
> >>>>>>>>>>>>>>>>>>>>>> custom stores would essentially lose
> >>>>>>> checkpointing
> >>>>>>>>>>> with no
> >>>>>>>>>>>>>>>>>> indication
> >>>>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>> they're expected to make changes, besides a
> >>>>>> line
> >>>>>>>> in
> >>>>>>>>>>> the
> >>>>>>>>>>>>>> release
> >>>>>>>>>>>>>>>>>>> notes. I
> >>>>>>>>>>>>>>>>>>>>>> agree that the best solution would be to
> >>>>>>> provide a
> >>>>>>>>>>> default
> >>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>> checkpoints
> >>>>>>>>>>>>>>>>>>>>>> to a file. The one thing I would change is
> >>>>>> that
> >>>>>>>> the
> >>>>>>>>>>>>>>> checkpointing
> >>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>>>>>> store-local file, instead of a per-Task file.
> >>>>>>> This
> >>>>>>>>>>> way the
> >>>>>>>>>>>>>>>>>> StateStore
> >>>>>>>>>>>>>>>>>>>>> still
> >>>>>>>>>>>>>>>>>>>>>> technically owns its own checkpointing (via a
> >>>>>>>>> default
> >>>>>>>>>>>>>>>>>> implementation),
> >>>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>> the StateManager/Task execution engine
> >>>>>> doesn't
> >>>>>>>> need
> >>>>>>>>>>> to know
> >>>>>>>>>>>>>>>>>> anything
> >>>>>>>>>>>>>>>>>>>>> about
> >>>>>>>>>>>>>>>>>>>>>> checkpointing, which greatly simplifies some
> >>>>>> of
> >>>>>>>> the
> >>>>>>>>>>> logic.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 2. OOME errors: The main reasons why I didn't
> >>>>>>>>> explore
> >>>>>>>>>>> a
> >>>>>>>>>>>>>>> solution
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>> this is
> >>>>>>>>>>>>>>>>>>>>>> a) to keep this KIP as simple as possible,
> >>>>>> and
> >>>>>>> b)
> >>>>>>>>>>> because
> >>>>>>>>>>>>> I'm
> >>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>> exactly
> >>>>>>>>>>>>>>>>>>>>>> how to signal that a Task should commit
> >>>>>>>> prematurely.
> >>>>>>>>>>> I'm
> >>>>>>>>>>>>>>>> confident
> >>>>>>>>>>>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>>>> possible, and I think it's worth adding a
> >>>>>>> section
> >>>>>>>> on
> >>>>>>>>>>>>> handling
> >>>>>>>>>>>>>>>> this.
> >>>>>>>>>>>>>>>>>>>>> Besides
> >>>>>>>>>>>>>>>>>>>>>> my proposal to force an early commit once
> >>>>>> memory
> >>>>>>>>> usage
> >>>>>>>>>>>>>> reaches
> >>>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>>>>> threshold,
> >>>>>>>>>>>>>>>>>>>>>> is there any other approach that you might
> >>>>>>> suggest
> >>>>>>>>> for
> >>>>>>>>>>>>>> tackling
> >>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>> problem?
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 3. ALOS: I can add in an explicit paragraph,
> >>>>>> but
> >>>>>>>> my
> >>>>>>>>>>>>>> assumption
> >>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>> since transactional behaviour comes at
> >>>>>> little/no
> >>>>>>>>>>> cost, that
> >>>>>>>>>>>>>> it
> >>>>>>>>>>>>>>>>>> should
> >>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>> available by default on all stores,
> >>>>>> irrespective
> >>>>>>>> of
> >>>>>>>>>>> the
> >>>>>>>>>>>>>>>> processing
> >>>>>>>>>>>>>>>>>>> mode.
> >>>>>>>>>>>>>>>>>>>>>> While ALOS doesn't use transactions, the Task
> >>>>>>>> itself
> >>>>>>>>>>> still
> >>>>>>>>>>>>>>>>>> "commits",
> >>>>>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>>>>> the behaviour should be correct under ALOS
> >>>>>> too.
> >>>>>>>> I'm
> >>>>>>>>>>> not
> >>>>>>>>>>>>>>> convinced
> >>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>>>> worth having both
> >>>>>>> transactional/non-transactional
> >>>>>>>>>>> stores
> >>>>>>>>>>>>>>>>>> available, as
> >>>>>>>>>>>>>>>>>>>>> it
> >>>>>>>>>>>>>>>>>>>>>> would considerably increase the complexity of
> >>>>>>> the
> >>>>>>>>>>> codebase,
> >>>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>> very
> >>>>>>>>>>>>>>>>>>>>> little
> >>>>>>>>>>>>>>>>>>>>>> benefit.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 4. Method deprecation: Are you referring to
> >>>>>>>>>>>>>>>>>> StateStore#getPosition()?
> >>>>>>>>>>>>>>>>>>>>> As I
> >>>>>>>>>>>>>>>>>>>>>> understand it, Position contains the
> >>>>>> position of
> >>>>>>>> the
> >>>>>>>>>>>>> *source*
> >>>>>>>>>>>>>>>>>> topics,
> >>>>>>>>>>>>>>>>>>>>>> whereas the commit offsets would be the
> >>>>>>>> *changelog*
> >>>>>>>>>>>>> offsets.
> >>>>>>>>>>>>>> So
> >>>>>>>>>>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>>> still
> >>>>>>>>>>>>>>>>>>>>>> necessary to retain the Position data, as
> >>>>>> well
> >>>>>>> as
> >>>>>>>>> the
> >>>>>>>>>>>>>> changelog
> >>>>>>>>>>>>>>>>>>> offsets.
> >>>>>>>>>>>>>>>>>>>>>> What I meant in the KIP is that Position
> >>>>>> offsets
> >>>>>>>> are
> >>>>>>>>>>>>>> currently
> >>>>>>>>>>>>>>>>>> stored
> >>>>>>>>>>>>>>>>>>>>> in a
> >>>>>>>>>>>>>>>>>>>>>> file, and since we can atomically store
> >>>>>> metadata
> >>>>>>>>>>> along with
> >>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>> record
> >>>>>>>>>>>>>>>>>>>>>> batch we commit to RocksDB, we can move our
> >>>>>>>> Position
> >>>>>>>>>>>>> offsets
> >>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>> metadata too, and gain the same transactional
> >>>>>>>>>>> guarantees
> >>>>>>>>>>>>> that
> >>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>>>>>> changelog offsets, ensuring that the Position
> >>>>>>>>> offsets
> >>>>>>>>>>> are
> >>>>>>>>>>>>>>>>>> consistent
> >>>>>>>>>>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>>>>>> the records that are read from the database.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> On Tue, 22 Nov 2022 at 16:25, John Roesler <
> >>>>>>>>>>>>>>> vvcep...@apache.org>
> >>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Thanks for publishing this alternative,
> >>>>>> Nick!
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> The benchmark you mentioned in the KIP-844
> >>>>>>>>>>> discussion
> >>>>>>>>>>>>> seems
> >>>>>>>>>>>>>>>> like
> >>>>>>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>>>>>>> compelling reason to revisit the built-in
> >>>>>>>>>>>>> transactionality
> >>>>>>>>>>>>>>>>>>> mechanism.
> >>>>>>>>>>>>>>>>>>>>> I
> >>>>>>>>>>>>>>>>>>>>>>> also appreciate you analysis, showing that
> >>>>>> for
> >>>>>>>>> most
> >>>>>>>>>>> use
> >>>>>>>>>>>>>>> cases,
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> write
> >>>>>>>>>>>>>>>>>>>>>>> batch approach should be just fine.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> There are a couple of points that would
> >>>>>> hold
> >>>>>>> me
> >>>>>>>>>>> back from
> >>>>>>>>>>>>>>>>>> approving
> >>>>>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>>> KIP right now:
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 1. Loss of coverage for custom stores.
> >>>>>>>>>>>>>>>>>>>>>>> The fact that you can plug in a
> >>>>>> (relatively)
> >>>>>>>>> simple
> >>>>>>>>>>>>>>>>>> implementation
> >>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> XStateStore interfaces and automagically
> >>>>>> get a
> >>>>>>>>>>>>> distributed
> >>>>>>>>>>>>>>>>>> database
> >>>>>>>>>>>>>>>>>>>>> out
> >>>>>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>>>> it is a significant benefit of Kafka
> >>>>>> Streams.
> >>>>>>>> I'd
> >>>>>>>>>>> hate to
> >>>>>>>>>>>>>>> lose
> >>>>>>>>>>>>>>>>>> it,
> >>>>>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>>>> it
> >>>>>>>>>>>>>>>>>>>>>>> would be better to spend some time and
> >>>>>> come up
> >>>>>>>>> with
> >>>>>>>>>>> a way
> >>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>> preserve
> >>>>>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>>> property. For example, can we provide a
> >>>>>>> default
> >>>>>>>>>>>>>>> implementation
> >>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>>>> `commit(..)` that re-implements the
> >>>>>> existing
> >>>>>>>>>>>>>> checkpoint-file
> >>>>>>>>>>>>>>>>>>>>> approach? Or
> >>>>>>>>>>>>>>>>>>>>>>> perhaps add an `isTransactional()` flag to
> >>>>>> the
> >>>>>>>>> state
> >>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>> interface
> >>>>>>>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>>>>>> that the runtime can decide whether to
> >>>>>>> continue
> >>>>>>>> to
> >>>>>>>>>>> manage
> >>>>>>>>>>>>>>>>>> checkpoint
> >>>>>>>>>>>>>>>>>>>>>> files
> >>>>>>>>>>>>>>>>>>>>>>> vs delegating transactionality to the
> >>>>>> stores?
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 2. Guarding against OOME
> >>>>>>>>>>>>>>>>>>>>>>> I appreciate your analysis, but I don't
> >>>>>> think
> >>>>>>>> it's
> >>>>>>>>>>>>>> sufficient
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>> say
> >>>>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>>> we will solve the memory problem later if
> >>>>>> it
> >>>>>>>>> becomes
> >>>>>>>>>>>>>>> necessary.
> >>>>>>>>>>>>>>>>>> The
> >>>>>>>>>>>>>>>>>>>>>>> experience leading to that situation would
> >>>>>> be
> >>>>>>>>> quite
> >>>>>>>>>>> bad:
> >>>>>>>>>>>>>>>> Imagine,
> >>>>>>>>>>>>>>>>>>> you
> >>>>>>>>>>>>>>>>>>>>>>> upgrade to AK 3.next, your tests pass, so
> >>>>>> you
> >>>>>>>>>>> deploy to
> >>>>>>>>>>>>>>>>>> production.
> >>>>>>>>>>>>>>>>>>>>> That
> >>>>>>>>>>>>>>>>>>>>>>> night, you get paged because your app is
> >>>>>> now
> >>>>>>>>>>> crashing
> >>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>> OOMEs. As
> >>>>>>>>>>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>>>>>>> all OOMEs, you'll have a really hard time
> >>>>>>>> finding
> >>>>>>>>>>> the
> >>>>>>>>>>>>> root
> >>>>>>>>>>>>>>>> cause,
> >>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>> once
> >>>>>>>>>>>>>>>>>>>>>>> you do, you won't have a clear path to
> >>>>>> resolve
> >>>>>>>> the
> >>>>>>>>>>> issue.
> >>>>>>>>>>>>>> You
> >>>>>>>>>>>>>>>>>> could
> >>>>>>>>>>>>>>>>>>>>> only
> >>>>>>>>>>>>>>>>>>>>>>> tune down the commit interval and cache
> >>>>>> buffer
> >>>>>>>>> size
> >>>>>>>>>>> until
> >>>>>>>>>>>>>> you
> >>>>>>>>>>>>>>>>>> stop
> >>>>>>>>>>>>>>>>>>>>>> getting
> >>>>>>>>>>>>>>>>>>>>>>> crashes.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> FYI, I know of multiple cases where people
> >>>>>> run
> >>>>>>>> EOS
> >>>>>>>>>>> with
> >>>>>>>>>>>>>> much
> >>>>>>>>>>>>>>>>>> larger
> >>>>>>>>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>>>> intervals to get better batching than the
> >>>>>>>> default,
> >>>>>>>>>>> so I
> >>>>>>>>>>>>>> don't
> >>>>>>>>>>>>>>>>>> think
> >>>>>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>>> pathological case would be as rare as you
> >>>>>>>> suspect.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Given that we already have the rudiments
> >>>>>> of an
> >>>>>>>>> idea
> >>>>>>>>>>> of
> >>>>>>>>>>>>> what
> >>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>> could
> >>>>>>>>>>>>>>>>>>>>> do
> >>>>>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>> prevent this downside, we should take the
> >>>>>> time
> >>>>>>>> to
> >>>>>>>>>>> design
> >>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>> solution.
> >>>>>>>>>>>>>>>>>>>>> We
> >>>>>>>>>>>>>>>>>>>>>> owe
> >>>>>>>>>>>>>>>>>>>>>>> it to our users to ensure that awesome new
> >>>>>>>>> features
> >>>>>>>>>>> don't
> >>>>>>>>>>>>>>> come
> >>>>>>>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>>>>>> bitter
> >>>>>>>>>>>>>>>>>>>>>>> pills unless we can't avoid it.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 3. ALOS mode.
> >>>>>>>>>>>>>>>>>>>>>>> On the other hand, I didn't see an
> >>>>>> indication
> >>>>>>> of
> >>>>>>>>> how
> >>>>>>>>>>>>> stores
> >>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>> handled under ALOS (aka non-EOS) mode.
> >>>>>>>>>>> Theoretically, the
> >>>>>>>>>>>>>>>>>>>>>> transactionality
> >>>>>>>>>>>>>>>>>>>>>>> of the store and the processing mode are
> >>>>>>>>>>> orthogonal. A
> >>>>>>>>>>>>>>>>>> transactional
> >>>>>>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>> would serve ALOS just as well as a
> >>>>>>>>>>> non-transactional one
> >>>>>>>>>>>>>> (if
> >>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>> better).
> >>>>>>>>>>>>>>>>>>>>>>> Under ALOS, though, the default commit
> >>>>>>> interval
> >>>>>>>> is
> >>>>>>>>>>> five
> >>>>>>>>>>>>>>>> minutes,
> >>>>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> memory issue is far more pressing.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> As I see it, we have several options to
> >>>>>>> resolve
> >>>>>>>>> this
> >>>>>>>>>>>>> point.
> >>>>>>>>>>>>>>> We
> >>>>>>>>>>>>>>>>>> could
> >>>>>>>>>>>>>>>>>>>>>>> demonstrate that transactional stores work
> >>>>>>> just
> >>>>>>>>>>> fine for
> >>>>>>>>>>>>>> ALOS
> >>>>>>>>>>>>>>>>>> and we
> >>>>>>>>>>>>>>>>>>>>> can
> >>>>>>>>>>>>>>>>>>>>>>> therefore just swap over unconditionally.
> >>>>>> We
> >>>>>>>> could
> >>>>>>>>>>> also
> >>>>>>>>>>>>>>> disable
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> transactional mechanism under ALOS so that
> >>>>>>>> stores
> >>>>>>>>>>> operate
> >>>>>>>>>>>>>>> just
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> same
> >>>>>>>>>>>>>>>>>>>>>> as
> >>>>>>>>>>>>>>>>>>>>>>> they do today when run in ALOS mode.
> >>>>>> Finally,
> >>>>>>> we
> >>>>>>>>>>> could do
> >>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>> same
> >>>>>>>>>>>>>>>>>>> as
> >>>>>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>>> KIP-844 and make transactional stores
> >>>>>> opt-in
> >>>>>>>> (it'd
> >>>>>>>>>>> be
> >>>>>>>>>>>>>> better
> >>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>> avoid
> >>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> extra opt-in mechanism, but it's a good
> >>>>>>>>>>>>>> get-out-of-jail-free
> >>>>>>>>>>>>>>>>>> card).
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 4. (minor point) Deprecation of methods
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> You mentioned that the new `commit` method
> >>>>>>>>> replaces
> >>>>>>>>>>>>> flush,
> >>>>>>>>>>>>>>>>>>>>>>> updateChangelogOffsets, and checkpoint. It
> >>>>>>> seems
> >>>>>>>>> to
> >>>>>>>>>>> me
> >>>>>>>>>>>>> that
> >>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>> point
> >>>>>>>>>>>>>>>>>>>>>> about
> >>>>>>>>>>>>>>>>>>>>>>> atomicity and Position also suggests that
> >>>>>> it
> >>>>>>>>>>> replaces the
> >>>>>>>>>>>>>>>>>> Position
> >>>>>>>>>>>>>>>>>>>>>>> callbacks. However, the proposal only
> >>>>>>> deprecates
> >>>>>>>>>>> `flush`.
> >>>>>>>>>>>>>>>> Should
> >>>>>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>> deprecating other methods as well?
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Thanks again for the KIP! It's really nice
> >>>>>>> that
> >>>>>>>>> you
> >>>>>>>>>>> and
> >>>>>>>>>>>>>> Alex
> >>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>>> get
> >>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> chance to collaborate on both directions so
> >>>>>>> that
> >>>>>>>>> we
> >>>>>>>>>>> can
> >>>>>>>>>>>>> get
> >>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>> best
> >>>>>>>>>>>>>>>>>>>>>>> outcome for Streams and its users.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> -John
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> On 2022/11/21 15:02:15 Nick Telford wrote:
> >>>>>>>>>>>>>>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> As I mentioned in the discussion thread
> >>>>>> for
> >>>>>>>>>>> KIP-844,
> >>>>>>>>>>>>> I've
> >>>>>>>>>>>>>>>> been
> >>>>>>>>>>>>>>>>>>>>> working
> >>>>>>>>>>>>>>>>>>>>>> on
> >>>>>>>>>>>>>>>>>>>>>>>> an alternative approach to achieving
> >>>>>> better
> >>>>>>>>>>>>> transactional
> >>>>>>>>>>>>>>>>>>> semantics
> >>>>>>>>>>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>>>>>>>> Kafka Streams StateStores.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> I've published this separately as
> >>>>>> KIP-892:
> >>>>>>>>>>>>> Transactional
> >>>>>>>>>>>>>>>>>> Semantics
> >>>>>>>>>>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>>>>>>>> StateStores
> >>>>>>>>>>>>>>>>>>>>>>>> <
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-892%3A+Transactional+Semantics+for+StateStores
> >>>>>>>>>>>>>>>>>>>>>>>> ,
> >>>>>>>>>>>>>>>>>>>>>>>> so that it can be discussed/reviewed
> >>>>>>>> separately
> >>>>>>>>>>> from
> >>>>>>>>>>>>>>> KIP-844.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> Alex: I'm especially interested in what
> >>>>>> you
> >>>>>>>>> think!
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> I have a nearly complete implementation
> >>>>>> of
> >>>>>>> the
> >>>>>>>>>>> changes
> >>>>>>>>>>>>>>>>>> outlined in
> >>>>>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>>>> KIP, please let me know if you'd like me
> >>>>>> to
> >>>>>>>> push
> >>>>>>>>>>> them
> >>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>> review
> >>>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>>> advance
> >>>>>>>>>>>>>>>>>>>>>>>> of a vote.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>
> >>>
> >>
> >
>
