I'd agree with you guys that as long as we are in agreement about the
configuration semantics, that would be a big win to move forward for
this KIP. As for the TaskCorruptedException handling like wiping state
stores, we can discuss that in the PR rather than in the KIP.
Just to clarify, I'm onboard with the latest proposal, and probably we
can move on for voting on this KIP now?
Guozhang
On Thu, Oct 19, 2023 at 5:33 AM Bruno Cadonna <cado...@apache.org> wrote:
>
> Hi Nick,
>
> What you and Lucas wrote about the different configurations of ALOS/EOS
> and READ_COMMITTED/READ_UNCOMMITTED make sense to me. My earlier
> concerns about changelogs diverging from the content of the local state
> stores turned out to not apply. So I think, we can move on with those
> configurations.
>
> Regarding the TaskCorruptedException and wiping out the state stores
> under EOS, couldn't we abort the transaction on the state store and
> close the task dirty? If the Kafka transaction was indeed committed, the
> store would restore the missing part from the changelog topic. If the
> Kafka transaction was not committed, changelog topic and state store are
> in-sync.
>
> In any case, IMO those are implementation details that we do not need to
> discuss and solve in the KIP discussion. We can solve them on the PR.
> The important thing is that the processing guarantees hold.
>
> Best,
> Bruno
>
> On 10/18/23 3:56 PM, Nick Telford wrote:
> > Hi Lucas,
> >
> > TaskCorruptedException is how Streams signals that the Task state needs to
> > be wiped, so we can't retain that exception without also wiping state on
> > timeouts.
> >
> > Regards,
> > Nick
> >
> > On Wed, 18 Oct 2023 at 14:48, Lucas Brutschy
> > <lbruts...@confluent.io.invalid>
> > wrote:
> >
> >> Hi Nick,
> >>
> >> I think indeed the better behavior would be to retry commitTransaction
> >> until we risk running out of time to meet `max.poll.interval.ms`.
> >>
> >> However, if it's handled as a `TaskCorruptedException` at the moment,
> >> I would do the same in this KIP, and leave exception handling
> >> improvements to future work. This KIP is already improving the
> >> situation a lot by not wiping the state store.
> >>
> >> Cheers,
> >> Lucas
> >>
> >> On Tue, Oct 17, 2023 at 3:51 PM Nick Telford <nick.telf...@gmail.com>
> >> wrote:
> >>>
> >>> Hi Lucas,
> >>>
> >>> Yeah, this is pretty much the direction I'm thinking of going in now. You
> >>> make an interesting point about committing on-error under
> >>> ALOS/READ_COMMITTED, although I haven't had a chance to think through the
> >>> implications yet.
> >>>
> >>> Something that I ran into earlier this week is an issue with the new
> >>> handling of TimeoutException. Without TX stores, TimeoutException under
> >> EOS
> >>> throws a TaskCorruptedException, which wipes the stores. However, with TX
> >>> stores, TimeoutException is now just bubbled up and dealt with as it is
> >>> under ALOS. The problem arises when the Producer#commitTransaction call
> >>> times out: Streams attempts to ignore the error and continue producing,
> >>> which causes the next call to Producer#send to throw
> >>> "IllegalStateException: Cannot attempt operation `send` because the
> >>> previous call to `commitTransaction` timed out and must be retried".
> >>>
> >>> I'm not sure what we should do here: retrying the commitTransaction seems
> >>> logical, but what if it times out again? Where do we draw the line and
> >>> shutdown the instance?
> >>>
> >>> Regards,
> >>> Nick
> >>>
> >>> On Mon, 16 Oct 2023 at 13:19, Lucas Brutschy <lbruts...@confluent.io
> >> .invalid>
> >>> wrote:
> >>>
> >>>> Hi all,
> >>>>
> >>>> I think I liked your suggestion of allowing EOS with READ_UNCOMMITTED,
> >>>> but keep wiping the state on error, and I'd vote for this solution
> >>>> when introducing `default.state.isolation.level`. This way, we'd have
> >>>> the most low-risk roll-out of this feature (no behavior change without
> >>>> reconfiguration), with the possibility of switching to the most sane /
> >>>> battle-tested default settings in 4.0. Essentially, we'd have a
> >>>> feature flag but call it `default.state.isolation.level` and don't
> >>>> have to deprecate it later.
> >>>>
> >>>> So the possible configurations would then be this:
> >>>>
> >>>> 1. ALOS/READ_UNCOMMITTED (default) = processing uses direct-to-DB, IQ
> >>>> reads from DB.
> >>>> 2. ALOS/READ_COMMITTED = processing uses WriteBatch, IQ reads from
> >>>> WriteBatch/DB. Flush on error (see note below).
> >>>> 3. EOS/READ_UNCOMMITTED (default) = processing uses direct-to-DB, IQ
> >>>> reads from DB. Wipe state on error.
> >>>> 4. EOS/READ_COMMITTED = processing uses WriteBatch, IQ reads from
> >>>> WriteBatch/DB.
> >>>>
> >>>> I believe the feature is important enough that we will see good
> >>>> adoption even without changing the default. In 4.0, when we have seen
> >>>> this being adopted and is battle-tested, we make READ_COMMITTED the
> >>>> default for EOS, or even READ_COMITTED always the default, depending
> >>>> on our experiences. And we could add a clever implementation of
> >>>> READ_UNCOMITTED with WriteBatches later.
> >>>>
> >>>> The only smell here is that `default.state.isolation.level` wouldn't
> >>>> be purely an IQ setting, but it would also (slightly) change the
> >>>> behavior of the processing, but that seems unavoidable as long as we
> >>>> haven't solve READ_UNCOMITTED IQ with WriteBatches.
> >>>>
> >>>> Minor: As for Bruno's point 4, I think if we are concerned about this
> >>>> behavior (we don't necessarily have to be, because it doesn't violate
> >>>> ALOS guarantees as far as I can see), we could make
> >>>> ALOS/READ_COMMITTED more similar to ALOS/READ_UNCOMITTED by flushing
> >>>> the WriteBatch on error (obviously, only if we have a chance to do
> >>>> that).
> >>>>
> >>>> Cheers,
> >>>> Lucas
> >>>>
> >>>> On Mon, Oct 16, 2023 at 12:19 PM Nick Telford <nick.telf...@gmail.com>
> >>>> wrote:
> >>>>>
> >>>>> Hi Guozhang,
> >>>>>
> >>>>> The KIP as it stands introduces a new configuration,
> >>>>> default.state.isolation.level, which is independent of
> >> processing.mode.
> >>>>> It's intended that this new configuration be used to configure a
> >> global
> >>>> IQ
> >>>>> isolation level in the short term, with a future KIP introducing the
> >>>>> capability to change the isolation level on a per-query basis,
> >> falling
> >>>> back
> >>>>> to the "default" defined by this config. That's why I called it
> >>>> "default",
> >>>>> for future-proofing.
> >>>>>
> >>>>> However, it currently includes the caveat that READ_UNCOMMITTED is
> >> not
> >>>>> available under EOS. I think this is the coupling you are alluding
> >> to?
> >>>>>
> >>>>> This isn't intended to be a restriction of the API, but is currently
> >> a
> >>>>> technical limitation. However, after discussing with some users about
> >>>>> use-cases that would require READ_UNCOMMITTED under EOS, I'm
> >> inclined to
> >>>>> remove that clause and put in the necessary work to make that
> >> combination
> >>>>> possible now.
> >>>>>
> >>>>> I currently see two possible approaches:
> >>>>>
> >>>>> 1. Disable TX StateStores internally when the IsolationLevel is
> >>>>> READ_UNCOMMITTED and the processing.mode is EOS. This is more
> >>>> difficult
> >>>>> than it sounds, as there are many assumptions being made
> >> throughout
> >>>> the
> >>>>> internals about the guarantees StateStores provide. It would
> >>>> definitely add
> >>>>> a lot of extra "if (read_uncommitted && eos)" branches,
> >> complicating
> >>>>> maintenance and testing.
> >>>>> 2. Invest the time *now* to make READ_UNCOMMITTED of EOS
> >> StateStores
> >>>>> possible. I have some ideas on how this could be achieved, but
> >> they
> >>>> would
> >>>>> need testing and could introduce some additional issues. The
> >> benefit
> >>>> of
> >>>>> this approach is that it would make query-time IsolationLevels
> >> much
> >>>> simpler
> >>>>> to implement in the future.
> >>>>>
> >>>>> Unfortunately, both will require considerable work that will further
> >>>> delay
> >>>>> this KIP, which was the reason I placed the restriction in the KIP
> >> in the
> >>>>> first place.
> >>>>>
> >>>>> Regards,
> >>>>> Nick
> >>>>>
> >>>>> On Sat, 14 Oct 2023 at 03:30, Guozhang Wang <
> >> guozhang.wang...@gmail.com>
> >>>>> wrote:
> >>>>>
> >>>>>> Hello Nick,
> >>>>>>
> >>>>>> First of all, thanks a lot for the great effort you've put in
> >> driving
> >>>>>> this KIP! I really like it coming through finally, as many people
> >> in
> >>>>>> the community have raised this. At the same time I honestly feel a
> >> bit
> >>>>>> ashamed for not putting enough of my time supporting it and
> >> pushing it
> >>>>>> through the finish line (you raised this KIP almost a year ago).
> >>>>>>
> >>>>>> I briefly passed through the DISCUSS thread so far, not sure I've
> >> 100
> >>>>>> percent digested all the bullet points. But with the goal of
> >> trying to
> >>>>>> help take it through the finish line in mind, I'd want to throw
> >>>>>> thoughts on top of my head only on the point #4 above which I felt
> >> may
> >>>>>> be the main hurdle for the current KIP to drive to a consensus now.
> >>>>>>
> >>>>>> The general question I asked myself is, whether we want to couple
> >> "IQ
> >>>>>> reading mode" with "processing mode". While technically I tend to
> >>>>>> agree with you that, it's feels like a bug if some single user
> >> chose
> >>>>>> "EOS" for processing mode while choosing "read uncommitted" for IQ
> >>>>>> reading mode, at the same time, I'm thinking if it's possible that
> >>>>>> there could be two different persons (or even two teams) that
> >> would be
> >>>>>> using the stream API to build the app, and the IQ API to query the
> >>>>>> running state of the app. I know this is less of a technical thing
> >> but
> >>>>>> rather a more design stuff, but if it could be ever the case, I'm
> >>>>>> wondering if the personale using the IQ API knows about the risks
> >> of
> >>>>>> using read uncommitted but still chose so for the favor of
> >>>>>> performance, no matter if the underlying stream processing mode
> >>>>>> configured by another personale is EOS or not. In that regard, I'm
> >>>>>> leaning towards a "leaving the door open, and close it later if we
> >>>>>> found it's a bad idea" aspect with a configuration that we can
> >>>>>> potentially deprecate than "shut the door, clean for everyone".
> >> More
> >>>>>> specifically, allowing the processing mode / IQ read mode to be
> >>>>>> decoupled, and if we found that there's no such cases as I
> >> speculated
> >>>>>> above or people started complaining a lot, we can still enforce
> >>>>>> coupling them.
> >>>>>>
> >>>>>> Again, just my 2c here. Thanks again for the great patience and
> >>>>>> diligence on this KIP.
> >>>>>>
> >>>>>>
> >>>>>> Guozhang
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>> On Fri, Oct 13, 2023 at 8:48 AM Nick Telford <
> >> nick.telf...@gmail.com>
> >>>>>> wrote:
> >>>>>>>
> >>>>>>> Hi Bruno,
> >>>>>>>
> >>>>>>> 4.
> >>>>>>> I'll hold off on making that change until we have a consensus as
> >> to
> >>>> what
> >>>>>>> configuration to use to control all of this, as it'll be
> >> affected by
> >>>> the
> >>>>>>> decision on EOS isolation levels.
> >>>>>>>
> >>>>>>> 5.
> >>>>>>> Done. I've chosen "committedOffsets".
> >>>>>>>
> >>>>>>> Regards,
> >>>>>>> Nick
> >>>>>>>
> >>>>>>> On Fri, 13 Oct 2023 at 16:23, Bruno Cadonna <cado...@apache.org>
> >>>> wrote:
> >>>>>>>
> >>>>>>>> Hi Nick,
> >>>>>>>>
> >>>>>>>> 1.
> >>>>>>>> Yeah, you are probably right that it does not make too much
> >> sense.
> >>>>>>>> Thanks for the clarification!
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> 4.
> >>>>>>>> Yes, sorry for the back and forth, but I think for the sake of
> >> the
> >>>> KIP
> >>>>>>>> it is better to let the ALOS behavior as it is for now due to
> >> the
> >>>>>>>> possible issues you would run into. Maybe we can find a
> >> solution
> >>>> in the
> >>>>>>>> future. Now the question returns to whether we really need
> >>>>>>>> default.state.isolation.level. Maybe the config could be the
> >>>> feature
> >>>>>>>> flag Sophie requested.
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> 5.
> >>>>>>>> There is a guideline in Kafka not to use the get prefix for
> >>>> getters (at
> >>>>>>>> least in the public API). Thus, could you please rename
> >>>>>>>>
> >>>>>>>> getCommittedOffset(TopicPartition partition) ->
> >>>>>>>> committedOffsetFor(TopicPartition partition)
> >>>>>>>>
> >>>>>>>> You can also propose an alternative to committedOffsetFor().
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> Best,
> >>>>>>>> Bruno
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> On 10/13/23 3:21 PM, Nick Telford wrote:
> >>>>>>>>> Hi Bruno,
> >>>>>>>>>
> >>>>>>>>> Thanks for getting back to me.
> >>>>>>>>>
> >>>>>>>>> 1.
> >>>>>>>>> I think this should be possible. Are you thinking of the
> >>>> situation
> >>>>>> where
> >>>>>>>> a
> >>>>>>>>> user may downgrade to a previous version of Kafka Streams? In
> >>>> that
> >>>>>> case,
> >>>>>>>>> sadly, the RocksDBStore would get wiped by the older version
> >> of
> >>>> Kafka
> >>>>>>>>> Streams anyway, because that version wouldn't understand the
> >>>> extra
> >>>>>> column
> >>>>>>>>> family (that holds offsets), so the missing Position file
> >> would
> >>>>>>>>> automatically get rebuilt when the store is rebuilt from the
> >>>>>> changelog.
> >>>>>>>>> Are there other situations than downgrade where a
> >> transactional
> >>>> store
> >>>>>>>> could
> >>>>>>>>> be replaced by a non-transactional one? I can't think of any.
> >>>>>>>>>
> >>>>>>>>> 2.
> >>>>>>>>> Ahh yes, the Test Plan - my Kryptonite! This section
> >> definitely
> >>>>>> needs to
> >>>>>>>> be
> >>>>>>>>> fleshed out. I'll work on that. How much detail do you need?
> >>>>>>>>>
> >>>>>>>>> 3.
> >>>>>>>>> See my previous email discussing this.
> >>>>>>>>>
> >>>>>>>>> 4.
> >>>>>>>>> Hmm, this is an interesting point. Are you suggesting that
> >> under
> >>>> ALOS
> >>>>>>>>> READ_COMMITTED should not be supported?
> >>>>>>>>>
> >>>>>>>>> Regards,
> >>>>>>>>> Nick
> >>>>>>>>>
> >>>>>>>>> On Fri, 13 Oct 2023 at 13:52, Bruno Cadonna <
> >> cado...@apache.org>
> >>>>>> wrote:
> >>>>>>>>>
> >>>>>>>>>> Hi Nick,
> >>>>>>>>>>
> >>>>>>>>>> I think the KIP is converging!
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> 1.
> >>>>>>>>>> I am wondering whether it makes sense to write the position
> >> file
> >>>>>> during
> >>>>>>>>>> close as we do for the checkpoint file, so that in case the
> >>>> state
> >>>>>> store
> >>>>>>>>>> is replaced with a non-transactional state store the
> >>>>>> non-transactional
> >>>>>>>>>> state store finds the position file. I think, this is not
> >>>> strictly
> >>>>>>>>>> needed, but would be a nice behavior instead of just
> >> deleting
> >>>> the
> >>>>>>>>>> position file.
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> 2.
> >>>>>>>>>> The test plan does not mention integration tests. Do you not
> >>>> need to
> >>>>>>>>>> extend existing ones and add new ones. Also for upgrading
> >> and
> >>>>>>>>>> downgrading you might need integration and/or system tests.
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> 3.
> >>>>>>>>>> I think Sophie made a point. Although, IQ reading from
> >>>> uncommitted
> >>>>>> data
> >>>>>>>>>> under EOS might be considered a bug by some people. Thus,
> >> your
> >>>> KIP
> >>>>>> would
> >>>>>>>>>> fix a bug rather than changing the intended behavior.
> >> However, I
> >>>>>> also
> >>>>>>>>>> see that a feature flag would help users that rely on this
> >> buggy
> >>>>>>>>>> behavior (at least until AK 4.0).
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> 4.
> >>>>>>>>>> This is related to the previous point. I assume that the
> >>>> difference
> >>>>>>>>>> between READ_COMMITTED and READ_UNCOMMITTED for ALOS is
> >> that in
> >>>> the
> >>>>>>>>>> former you enable transactions on the state store and in the
> >>>> latter
> >>>>>> you
> >>>>>>>>>> disable them. If my assumption is correct, I think that is
> >> an
> >>>> issue.
> >>>>>>>>>> Let's assume under ALOS Streams fails over a couple of times
> >>>> more or
> >>>>>>>>>> less at the same step in processing after value 3 is added
> >> to an
> >>>>>>>>>> aggregation but the offset of the corresponding input record
> >>>> was not
> >>>>>>>>>> committed. Without transactions disabled, the aggregation
> >> value
> >>>>>> would
> >>>>>>>>>> increase by 3 for each failover. With transactions enabled,
> >>>> value 3
> >>>>>>>>>> would only be added to the aggregation once when the offset
> >> of
> >>>> the
> >>>>>> input
> >>>>>>>>>> record is committed and the transaction finally completes.
> >> So
> >>>> the
> >>>>>>>>>> content of the state store would change depending on the
> >>>>>> configuration
> >>>>>>>>>> for IQ. IMO, the content of the state store should be
> >>>> independent
> >>>>>> from
> >>>>>>>>>> IQ. Given this issue, I propose to not use transactions with
> >>>> ALOS at
> >>>>>>>>>> all. I was a big proponent of using transactions with ALOS,
> >> but
> >>>> I
> >>>>>>>>>> realized that transactions with ALOS is not as easy as
> >> enabling
> >>>>>>>>>> transactions on state stores. Another aspect that is
> >>>> problematic is
> >>>>>> that
> >>>>>>>>>> the changelog topic which actually replicates the state
> >> store
> >>>> is not
> >>>>>>>>>> transactional under ALOS. Thus, it might happen that the
> >> state
> >>>>>> store and
> >>>>>>>>>> the changelog differ in their content. All of this is maybe
> >>>> solvable
> >>>>>>>>>> somehow, but for the sake of this KIP, I would leave it for
> >> the
> >>>>>> future.
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> Best,
> >>>>>>>>>> Bruno
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> On 10/12/23 10:32 PM, Sophie Blee-Goldman wrote:
> >>>>>>>>>>> Hey Nick! First of all thanks for taking up this awesome
> >>>> feature,
> >>>>>> I'm
> >>>>>>>>>> sure
> >>>>>>>>>>> every single
> >>>>>>>>>>> Kafka Streams user and dev would agree that it is sorely
> >>>> needed.
> >>>>>>>>>>>
> >>>>>>>>>>> I've just been catching up on the KIP and surrounding
> >>>> discussion,
> >>>>>> so
> >>>>>>>>>> please
> >>>>>>>>>>> forgive me
> >>>>>>>>>>> for any misunderstandings or misinterpretations of the
> >> current
> >>>>>> plan and
> >>>>>>>>>>> don't hesitate to
> >>>>>>>>>>> correct me.
> >>>>>>>>>>>
> >>>>>>>>>>> Before I jump in, I just want to say that having seen this
> >>>> drag on
> >>>>>> for
> >>>>>>>> so
> >>>>>>>>>>> long, my singular
> >>>>>>>>>>> goal in responding is to help this KIP past a perceived
> >>>> impasse so
> >>>>>> we
> >>>>>>>> can
> >>>>>>>>>>> finally move on
> >>>>>>>>>>> to voting and implementing it. Long discussions are to be
> >>>> expected
> >>>>>> for
> >>>>>>>>>>> major features like
> >>>>>>>>>>> this but it's completely on us as the Streams devs to make
> >> sure
> >>>>>> there
> >>>>>>>> is
> >>>>>>>>>> an
> >>>>>>>>>>> end in sight
> >>>>>>>>>>> for any ongoing discussion.
> >>>>>>>>>>>
> >>>>>>>>>>> With that said, it's my understanding that the KIP as
> >> currently
> >>>>>>>> proposed
> >>>>>>>>>> is
> >>>>>>>>>>> just not tenable
> >>>>>>>>>>> for Kafka Streams, and would prevent some EOS users from
> >>>> upgrading
> >>>>>> to
> >>>>>>>> the
> >>>>>>>>>>> version it
> >>>>>>>>>>> first appears in. Given that we can't predict or guarantee
> >>>> whether
> >>>>>> any
> >>>>>>>> of
> >>>>>>>>>>> the followup KIPs
> >>>>>>>>>>> would be completed in the same release cycle as this one,
> >> we
> >>>> need
> >>>>>> to
> >>>>>>>> make
> >>>>>>>>>>> sure that the
> >>>>>>>>>>> feature is either compatible with all current users or else
> >>>>>>>>>> feature-flagged
> >>>>>>>>>>> so that they may
> >>>>>>>>>>> opt in/out.
> >>>>>>>>>>>
> >>>>>>>>>>> Therefore, IIUC we need to have either (or both) of these
> >> as
> >>>>>>>>>>> fully-implemented config options:
> >>>>>>>>>>> 1. default.state.isolation.level
> >>>>>>>>>>> 2. enable.transactional.state.stores
> >>>>>>>>>>>
> >>>>>>>>>>> This way EOS users for whom read_committed semantics are
> >> not
> >>>>>> viable can
> >>>>>>>>>>> still upgrade,
> >>>>>>>>>>> and either use the isolation.level config to leverage the
> >> new
> >>>> txn
> >>>>>> state
> >>>>>>>>>>> stores without sacrificing
> >>>>>>>>>>> their application semantics, or else simply keep the
> >>>> transactional
> >>>>>>>> state
> >>>>>>>>>>> stores disabled until we
> >>>>>>>>>>> are able to fully implement the isolation level
> >> configuration
> >>>> at
> >>>>>> either
> >>>>>>>>>> an
> >>>>>>>>>>> application or query level.
> >>>>>>>>>>>
> >>>>>>>>>>> Frankly you are the expert here and know much more about
> >> the
> >>>>>> tradeoffs
> >>>>>>>> in
> >>>>>>>>>>> both semantics and
> >>>>>>>>>>> effort level of implementing one of these configs vs the
> >>>> other. In
> >>>>>> my
> >>>>>>>>>>> opinion, either option would
> >>>>>>>>>>> be fine and I would leave the decision of which one to
> >> include
> >>>> in
> >>>>>> this
> >>>>>>>>>> KIP
> >>>>>>>>>>> completely up to you.
> >>>>>>>>>>> I just don't see a way for the KIP to proceed without some
> >>>>>> variation of
> >>>>>>>>>> the
> >>>>>>>>>>> above that would allow
> >>>>>>>>>>> EOS users to opt-out of read_committed.
> >>>>>>>>>>>
> >>>>>>>>>>> (If it's all the same to you, I would recommend always
> >>>> including a
> >>>>>>>>>> feature
> >>>>>>>>>>> flag in large structural
> >>>>>>>>>>> changes like this. No matter how much I trust someone or
> >>>> myself to
> >>>>>>>>>>> implement a feature, you just
> >>>>>>>>>>> never know what kind of bugs might slip in, especially
> >> with the
> >>>>>> very
> >>>>>>>>>> first
> >>>>>>>>>>> iteration that gets released.
> >>>>>>>>>>> So personally, my choice would be to add the feature flag
> >> and
> >>>>>> leave it
> >>>>>>>>>> off
> >>>>>>>>>>> by default. If all goes well
> >>>>>>>>>>> you can do a quick KIP to enable it by default as soon as
> >> the
> >>>>>>>>>>> isolation.level config has been
> >>>>>>>>>>> completed. But feel free to just pick whichever option is
> >>>> easiest
> >>>>>> or
> >>>>>>>>>>> quickest for you to implement)
> >>>>>>>>>>>
> >>>>>>>>>>> Hope this helps move the discussion forward,
> >>>>>>>>>>> Sophie
> >>>>>>>>>>>
> >>>>>>>>>>> On Tue, Sep 19, 2023 at 1:57 AM Nick Telford <
> >>>>>> nick.telf...@gmail.com>
> >>>>>>>>>> wrote:
> >>>>>>>>>>>
> >>>>>>>>>>>> Hi Bruno,
> >>>>>>>>>>>>
> >>>>>>>>>>>> Agreed, I can live with that for now.
> >>>>>>>>>>>>
> >>>>>>>>>>>> In an effort to keep the scope of this KIP from
> >> expanding, I'm
> >>>>>> leaning
> >>>>>>>>>>>> towards just providing a configurable
> >>>>>> default.state.isolation.level
> >>>>>>>> and
> >>>>>>>>>>>> removing IsolationLevel from the StateStoreContext. This
> >>>> would be
> >>>>>>>>>>>> compatible with adding support for query-time
> >> IsolationLevels
> >>>> in
> >>>>>> the
> >>>>>>>>>>>> future, whilst providing a way for users to select an
> >>>> isolation
> >>>>>> level
> >>>>>>>>>> now.
> >>>>>>>>>>>>
> >>>>>>>>>>>> The big problem with this, however, is that if a user
> >> selects
> >>>>>>>>>>>> processing.mode
> >>>>>>>>>>>> = "exactly-once(-v2|-beta)", and
> >>>> default.state.isolation.level =
> >>>>>>>>>>>> "READ_UNCOMMITTED", we need to guarantee that the data
> >> isn't
> >>>>>> written
> >>>>>>>> to
> >>>>>>>>>>>> disk until commit() is called, but we also need to permit
> >> IQ
> >>>>>> threads
> >>>>>>>> to
> >>>>>>>>>>>> read from the ongoing transaction.
> >>>>>>>>>>>>
> >>>>>>>>>>>> A simple solution would be to (temporarily) forbid this
> >>>>>> combination of
> >>>>>>>>>>>> configuration, and have default.state.isolation.level
> >>>>>> automatically
> >>>>>>>>>> switch
> >>>>>>>>>>>> to READ_COMMITTED when processing.mode is anything other
> >> than
> >>>>>>>>>>>> at-least-once. Do you think this would be acceptable?
> >>>>>>>>>>>>
> >>>>>>>>>>>> In a later KIP, we can add support for query-time
> >> isolation
> >>>>>> levels and
> >>>>>>>>>>>> solve this particular problem there, which would relax
> >> this
> >>>>>>>> restriction.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Regards,
> >>>>>>>>>>>> Nick
> >>>>>>>>>>>>
> >>>>>>>>>>>> On Tue, 19 Sept 2023 at 09:30, Bruno Cadonna <
> >>>> cado...@apache.org>
> >>>>>>>>>> wrote:
> >>>>>>>>>>>>
> >>>>>>>>>>>>> Why do we need to add READ_COMMITTED to
> >>>> InMemoryKeyValueStore? I
> >>>>>>>> think
> >>>>>>>>>>>>> it is perfectly valid to say InMemoryKeyValueStore do not
> >>>> support
> >>>>>>>>>>>>> READ_COMMITTED for now, since READ_UNCOMMITTED is the
> >>>> de-facto
> >>>>>>>> default
> >>>>>>>>>>>>> at the moment.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Best,
> >>>>>>>>>>>>> Bruno
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> On 9/18/23 7:12 PM, Nick Telford wrote:
> >>>>>>>>>>>>>> Oh! One other concern I haven't mentioned: if we make
> >>>>>>>> IsolationLevel a
> >>>>>>>>>>>>>> query-time constraint, then we need to add support for
> >>>>>>>> READ_COMMITTED
> >>>>>>>>>>>> to
> >>>>>>>>>>>>>> InMemoryKeyValueStore too, which will require some
> >> changes
> >>>> to
> >>>>>> the
> >>>>>>>>>>>>>> implementation.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> On Mon, 18 Sept 2023 at 17:24, Nick Telford <
> >>>>>> nick.telf...@gmail.com
> >>>>>>>>>
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> I agree that having IsolationLevel be determined at
> >>>> query-time
> >>>>>> is
> >>>>>>>> the
> >>>>>>>>>>>>>>> ideal design, but there are a few sticking points:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> 1.
> >>>>>>>>>>>>>>> There needs to be some way to communicate the
> >>>> IsolationLevel
> >>>>>> down
> >>>>>>>> to
> >>>>>>>>>>>> the
> >>>>>>>>>>>>>>> RocksDBStore itself, so that the query can respect it.
> >>>> Since
> >>>>>> stores
> >>>>>>>>>>>> are
> >>>>>>>>>>>>>>> "layered" in functionality (i.e. ChangeLoggingStore,
> >>>>>> MeteredStore,
> >>>>>>>>>>>>> etc.),
> >>>>>>>>>>>>>>> we need some way to deliver that information to the
> >> bottom
> >>>>>> layer.
> >>>>>>>> For
> >>>>>>>>>>>>> IQv2,
> >>>>>>>>>>>>>>> we can use the existing State#query() method, but IQv1
> >> has
> >>>> no
> >>>>>> way
> >>>>>>>> to
> >>>>>>>>>>>> do
> >>>>>>>>>>>>>>> this.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> A simple approach, which would potentially open up
> >> other
> >>>>>> options,
> >>>>>>>>>>>> would
> >>>>>>>>>>>>> be
> >>>>>>>>>>>>>>> to add something like: ReadOnlyKeyValueStore<K, V>
> >>>>>>>>>>>>>>> readOnlyView(IsolationLevel isolationLevel) to
> >>>>>>>> ReadOnlyKeyValueStore
> >>>>>>>>>>>>> (and
> >>>>>>>>>>>>>>> similar to ReadOnlyWindowStore, ReadOnlySessionStore,
> >>>> etc.).
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> 2.
> >>>>>>>>>>>>>>> As mentioned above, RocksDB WriteBatches are not
> >>>> thread-safe,
> >>>>>> which
> >>>>>>>>>>>>> causes
> >>>>>>>>>>>>>>> a problem if we want to provide READ_UNCOMMITTED
> >>>> Iterators. I
> >>>>>> also
> >>>>>>>>>>>> had a
> >>>>>>>>>>>>>>> look at RocksDB Transactions[1], but they solve a very
> >>>>>> different
> >>>>>>>>>>>>> problem,
> >>>>>>>>>>>>>>> and have the same thread-safety issue.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> One possible approach that I mentioned is chaining
> >>>>>> WriteBatches:
> >>>>>>>>>> every
> >>>>>>>>>>>>>>> time a new Interactive Query is received (i.e.
> >>>> readOnlyView,
> >>>>>> see
> >>>>>>>>>>>> above,
> >>>>>>>>>>>>>>> is called) we "freeze" the existing WriteBatch, and
> >> start a
> >>>>>> new one
> >>>>>>>>>>>> for
> >>>>>>>>>>>>> new
> >>>>>>>>>>>>>>> writes. The Interactive Query queries the "chain" of
> >>>> previous
> >>>>>>>>>>>>> WriteBatches
> >>>>>>>>>>>>>>> + the underlying database; while the StreamThread
> >> starts
> >>>>>> writing to
> >>>>>>>>>>>> the
> >>>>>>>>>>>>>>> *new* WriteBatch. On-commit, the StreamThread would
> >> write
> >>>> *all*
> >>>>>>>>>>>>>>> WriteBatches in the chain to the database (that have
> >> not
> >>>> yet
> >>>>>> been
> >>>>>>>>>>>>> written).
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> WriteBatches would be closed/freed only when they have
> >> been
> >>>>>> both
> >>>>>>>>>>>>>>> committed, and all open Interactive Queries on them
> >> have
> >>>> been
> >>>>>>>> closed.
> >>>>>>>>>>>>> This
> >>>>>>>>>>>>>>> would require some reference counting.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Obviously a drawback of this approach is the potential
> >> for
> >>>>>>>> increased
> >>>>>>>>>>>>>>> memory usage: if an Interactive Query is long-lived,
> >> for
> >>>>>> example by
> >>>>>>>>>>>>> doing a
> >>>>>>>>>>>>>>> full scan over a large database, or even just pausing
> >> in
> >>>> the
> >>>>>> middle
> >>>>>>>>>> of
> >>>>>>>>>>>>> an
> >>>>>>>>>>>>>>> iteration, then the existing chain of WriteBatches
> >> could be
> >>>>>> kept
> >>>>>>>>>>>> around
> >>>>>>>>>>>>> for
> >>>>>>>>>>>>>>> a long time, potentially forever.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> --
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> A.
> >>>>>>>>>>>>>>> Going off on a tangent, it looks like in addition to
> >>>> supporting
> >>>>>>>>>>>>>>> READ_COMMITTED queries, we could go further and support
> >>>>>>>>>>>> REPEATABLE_READ
> >>>>>>>>>>>>>>> queries (i.e. where subsequent reads to the same key
> >> in the
> >>>>>> same
> >>>>>>>>>>>>>>> Interactive Query are guaranteed to yield the same
> >> value)
> >>>> by
> >>>>>> making
> >>>>>>>>>>>> use
> >>>>>>>>>>>>> of
> >>>>>>>>>>>>>>> RocksDB Snapshots[2]. These are fairly lightweight, so
> >> the
> >>>>>>>>>> performance
> >>>>>>>>>>>>>>> impact is likely to be negligible, but they do require
> >>>> that the
> >>>>>>>>>>>>> Interactive
> >>>>>>>>>>>>>>> Query session can be explicitly closed.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> This could be achieved if we made the above
> >> readOnlyView
> >>>>>> interface
> >>>>>>>>>>>> look
> >>>>>>>>>>>>>>> more like:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> interface ReadOnlyKeyValueView<K, V> implements
> >>>>>>>>>>>> ReadOnlyKeyValueStore<K,
> >>>>>>>>>>>>>>> V>, AutoCloseable {}
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> interface ReadOnlyKeyValueStore<K, V> {
> >>>>>>>>>>>>>>> ...
> >>>>>>>>>>>>>>> ReadOnlyKeyValueView<K, V>
> >>>> readOnlyView(IsolationLevel
> >>>>>>>>>>>>> isolationLevel);
> >>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> But this would be a breaking change, as existing IQv1
> >>>> queries
> >>>>>> are
> >>>>>>>>>>>>>>> guaranteed to never call store.close(), and therefore
> >> these
> >>>>>> would
> >>>>>>>>>> leak
> >>>>>>>>>>>>>>> memory under REPEATABLE_READ.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> B.
> >>>>>>>>>>>>>>> One thing that's notable: MyRocks states that they
> >> support
> >>>>>>>>>>>>> READ_COMMITTED
> >>>>>>>>>>>>>>> and REPEATABLE_READ, but they make no mention of
> >>>>>>>>>>>> READ_UNCOMMITTED[3][4].
> >>>>>>>>>>>>>>> This could be because doing so is technically
> >>>>>> difficult/impossible
> >>>>>>>>>>>> using
> >>>>>>>>>>>>>>> the primitives available in RocksDB.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> --
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Lucas, to address your points:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> U1.
> >>>>>>>>>>>>>>> It's only "SHOULD" to permit alternative (i.e.
> >> non-RocksDB)
> >>>>>>>>>>>>>>> implementations of StateStore that do not support
> >> atomic
> >>>>>> writes.
> >>>>>>>>>>>>> Obviously
> >>>>>>>>>>>>>>> in those cases, the guarantees Kafka Streams
> >>>> provides/expects
> >>>>>> would
> >>>>>>>>>> be
> >>>>>>>>>>>>>>> relaxed. Do you think we should require all
> >>>> implementations to
> >>>>>>>>>> support
> >>>>>>>>>>>>>>> atomic writes?
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> U2.
> >>>>>>>>>>>>>>> Stores can support multiple IsolationLevels. As we've
> >>>> discussed
> >>>>>>>>>> above,
> >>>>>>>>>>>>> the
> >>>>>>>>>>>>>>> ideal scenario would be to specify the IsolationLevel
> >> at
> >>>>>>>> query-time.
> >>>>>>>>>>>>>>> Failing that, I think the second-best approach is to
> >>>> define the
> >>>>>>>>>>>>>>> IsolationLevel for *all* queries based on the
> >>>> processing.mode,
> >>>>>>>> which
> >>>>>>>>>>>> is
> >>>>>>>>>>>>>>> what the default StateStoreContext#isolationLevel()
> >>>> achieves.
> >>>>>> Would
> >>>>>>>>>>>> you
> >>>>>>>>>>>>>>> prefer an alternative?
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> While the existing implementation is equivalent to
> >>>>>>>> READ_UNCOMMITTED,
> >>>>>>>>>>>>> this
> >>>>>>>>>>>>>>> can yield unexpected results/errors under EOS, if a
> >>>>>> transaction is
> >>>>>>>>>>>>> rolled
> >>>>>>>>>>>>>>> back. While this would be a change in behaviour for
> >> users,
> >>>> it
> >>>>>> would
> >>>>>>>>>>>> look
> >>>>>>>>>>>>>>> more like a bug fix than a breaking change. That said,
> >> we
> >>>>>> *could*
> >>>>>>>>>> make
> >>>>>>>>>>>>> it
> >>>>>>>>>>>>>>> configurable, and default to the existing behaviour
> >>>>>>>>>> (READ_UNCOMMITTED)
> >>>>>>>>>>>>>>> instead of inferring it from the processing.mode?
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> N1, N2.
> >>>>>>>>>>>>>>> These were only primitives to avoid boxing costs, but
> >> since
> >>>>>> this is
> >>>>>>>>>>>> not
> >>>>>>>>>>>>> a
> >>>>>>>>>>>>>>> performance sensitive area, it should be fine to
> >> change if
> >>>>>> that's
> >>>>>>>>>>>>> desirable.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> N3.
> >>>>>>>>>>>>>>> It's because the store "manages its own offsets", which
> >>>>>> includes
> >>>>>>>> both
> >>>>>>>>>>>>>>> committing the offset, *and providing it* via
> >>>>>> getCommittedOffset().
> >>>>>>>>>>>>>>> Personally, I think "managesOffsets" conveys this best,
> >>>> but I
> >>>>>> don't
> >>>>>>>>>>>> mind
> >>>>>>>>>>>>>>> changing it if the nomenclature is unclear.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Sorry for the massive emails/essays!
> >>>>>>>>>>>>>>> --
> >>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> 1:
> >> https://github.com/facebook/rocksdb/wiki/Transactions
> >>>>>>>>>>>>>>> 2: https://github.com/facebook/rocksdb/wiki/Snapshot
> >>>>>>>>>>>>>>> 3:
> >>>>>>>>
> >> https://github.com/facebook/mysql-5.6/wiki/Transaction-Isolation
> >>>>>>>>>>>>>>> 4:
> >>>> https://mariadb.com/kb/en/myrocks-transactional-isolation/
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> On Mon, 18 Sept 2023 at 16:19, Lucas Brutschy
> >>>>>>>>>>>>>>> <lbruts...@confluent.io.invalid> wrote:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Hi Nick,
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> since I last read it in April, the KIP has become much
> >>>>>> cleaner and
> >>>>>>>>>>>>>>>> easier to read. Great work!
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> It feels to me the last big open point is whether we
> >> can
> >>>>>> implement
> >>>>>>>>>>>>>>>> isolation level as a query parameter. I understand
> >> that
> >>>> there
> >>>>>> are
> >>>>>>>>>>>>>>>> implementation concerns, but as Colt says, it would
> >> be a
> >>>> great
> >>>>>>>>>>>>>>>> addition, and would also simplify the migration path
> >> for
> >>>> this
> >>>>>>>>>> change.
> >>>>>>>>>>>>>>>> Is the implementation problem you mentioned caused by
> >> the
> >>>>>>>> WriteBatch
> >>>>>>>>>>>>>>>> not having a notion of a snapshot, as the underlying
> >> DB
> >>>>>> iterator
> >>>>>>>>>>>> does?
> >>>>>>>>>>>>>>>> In that case, I am not sure a chain of WriteBatches
> >> as you
> >>>>>> propose
> >>>>>>>>>>>>>>>> would fully solve the problem, but maybe I didn't dig
> >>>> enough
> >>>>>> into
> >>>>>>>>>> the
> >>>>>>>>>>>>>>>> details to fully understand it.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> If it's not possible to implement it now, would it be
> >> an
> >>>>>> option to
> >>>>>>>>>>>>>>>> make sure in this KIP that we do not fully close the
> >> door
> >>>> on
> >>>>>>>>>>>> per-query
> >>>>>>>>>>>>>>>> isolation levels in the interface, as it may be
> >> possible
> >>>> to
> >>>>>>>>>> implement
> >>>>>>>>>>>>>>>> the missing primitives in RocksDB or Speedb in the
> >> future.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Understanding:
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> * U1) Why is it only "SHOULD" for changelogOffsets to
> >> be
> >>>>>> persisted
> >>>>>>>>>>>>>>>> atomically with the records?
> >>>>>>>>>>>>>>>> * U2) Don't understand the default implementation of
> >>>>>>>>>>>> `isolationLevel`.
> >>>>>>>>>>>>>>>> The isolation level should be a property of the
> >> underlying
> >>>>>> store,
> >>>>>>>>>> and
> >>>>>>>>>>>>>>>> not be defined by the default config? Existing stores
> >>>> probably
> >>>>>>>> don't
> >>>>>>>>>>>>>>>> guarantee READ_COMMITTED, so the default should be to
> >>>> return
> >>>>>>>>>>>>>>>> READ_UNCOMMITTED.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Nits:
> >>>>>>>>>>>>>>>> * N1) Could `getComittedOffset` use an `OptionalLong`
> >>>> return
> >>>>>> type,
> >>>>>>>>>> to
> >>>>>>>>>>>>>>>> avoid the `null`?
> >>>>>>>>>>>>>>>> * N2) Could `apporixmateNumUncomittedBytes` use an
> >>>>>> `OptionalLong`
> >>>>>>>>>>>>>>>> return type, to avoid the `-1`?
> >>>>>>>>>>>>>>>> * N3) I don't understand why `managesOffsets` uses the
> >>>>>> 'manage'
> >>>>>>>>>> verb,
> >>>>>>>>>>>>>>>> whereas all other methods use the "commits" verb. I'd
> >>>> suggest
> >>>>>>>>>>>>>>>> `commitsOffsets`.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Either way, it feels this KIP is very close to the
> >> finish
> >>>>>> line,
> >>>>>>>> I'm
> >>>>>>>>>>>>>>>> looking forward to seeing this in production!
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Cheers,
> >>>>>>>>>>>>>>>> Lucas
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> On Mon, Sep 18, 2023 at 6:57 AM Colt McNealy <
> >>>>>> c...@littlehorse.io
> >>>>>>>>>
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> Making IsolationLevel a query-time constraint,
> >> rather
> >>>> than
> >>>>>>>> linking
> >>>>>>>>>>>> it
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>> the processing.guarantee.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> As I understand it, would this allow even a user of
> >> EOS
> >>>> to
> >>>>>>>> control
> >>>>>>>>>>>>>>>> whether
> >>>>>>>>>>>>>>>>> reading committed or uncommitted records? If so, I am
> >>>> highly
> >>>>>> in
> >>>>>>>>>>>> favor
> >>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>> this.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> I know that I was one of the early people to point
> >> out
> >>>> the
> >>>>>>>> current
> >>>>>>>>>>>>>>>>> shortcoming that IQ reads uncommitted records, but
> >> just
> >>>> this
> >>>>>>>>>>>> morning I
> >>>>>>>>>>>>>>>>> realized a pattern we use which means that (for
> >> certain
> >>>>>> queries)
> >>>>>>>>>> our
> >>>>>>>>>>>>>>>> system
> >>>>>>>>>>>>>>>>> needs to be able to read uncommitted records, which
> >> is
> >>>> the
> >>>>>>>> current
> >>>>>>>>>>>>>>>> behavior
> >>>>>>>>>>>>>>>>> of Kafka Streams in EOS.***
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> If IsolationLevel being a query-time decision allows
> >> for
> >>>>>> this,
> >>>>>>>> then
> >>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>> would be amazing. I would also vote that the default
> >>>> behavior
> >>>>>>>>>> should
> >>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>> reading uncommitted records, because it is totally
> >>>> possible
> >>>>>> for a
> >>>>>>>>>>>>> valid
> >>>>>>>>>>>>>>>>> application to depend on that behavior, and breaking
> >> it
> >>>> in a
> >>>>>>>> minor
> >>>>>>>>>>>>>>>> release
> >>>>>>>>>>>>>>>>> might be a bit strong.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> *** (Note, for the curious reader....) Our
> >> use-case/query
> >>>>>> pattern
> >>>>>>>>>>>> is a
> >>>>>>>>>>>>>>>> bit
> >>>>>>>>>>>>>>>>> complex, but reading "uncommitted" records is
> >> actually
> >>>> safe
> >>>>>> in
> >>>>>>>> our
> >>>>>>>>>>>>> case
> >>>>>>>>>>>>>>>>> because processing is deterministic. Additionally, IQ
> >>>> being
> >>>>>> able
> >>>>>>>> to
> >>>>>>>>>>>>> read
> >>>>>>>>>>>>>>>>> uncommitted records is crucial to enable "read your
> >> own
> >>>>>> writes"
> >>>>>>>> on
> >>>>>>>>>>>> our
> >>>>>>>>>>>>>>>> API:
> >>>>>>>>>>>>>>>>> Due to the deterministic processing, we send an
> >> "ack" to
> >>>> the
> >>>>>>>> client
> >>>>>>>>>>>>> who
> >>>>>>>>>>>>>>>>> makes the request as soon as the processor processes
> >> the
> >>>>>> result.
> >>>>>>>> If
> >>>>>>>>>>>>> they
> >>>>>>>>>>>>>>>>> can't read uncommitted records, they may receive a
> >> "201 -
> >>>>>>>> Created"
> >>>>>>>>>>>>>>>>> response, immediately followed by a "404 - Not Found"
> >>>> when
> >>>>>> doing
> >>>>>>>> a
> >>>>>>>>>>>>>>>> lookup
> >>>>>>>>>>>>>>>>> for the object they just created).
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Thanks,
> >>>>>>>>>>>>>>>>> Colt McNealy
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> *Founder, LittleHorse.dev*
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> On Wed, Sep 13, 2023 at 9:19 AM Nick Telford <
> >>>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> Addendum:
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> I think we would also face the same problem with the
> >>>>>> approach
> >>>>>>>> John
> >>>>>>>>>>>>>>>> outlined
> >>>>>>>>>>>>>>>>>> earlier (using the record cache as a transaction
> >> buffer
> >>>> and
> >>>>>>>>>>>> flushing
> >>>>>>>>>>>>>>>> it
> >>>>>>>>>>>>>>>>>> straight to SST files). This is because the record
> >> cache
> >>>>>> (the
> >>>>>>>>>>>>>>>> ThreadCache
> >>>>>>>>>>>>>>>>>> class) is not thread-safe, so every commit would
> >>>> invalidate
> >>>>>> open
> >>>>>>>>>> IQ
> >>>>>>>>>>>>>>>>>> Iterators in the same way that RocksDB WriteBatches
> >> do.
> >>>>>>>>>>>>>>>>>> --
> >>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>> On Wed, 13 Sept 2023 at 16:58, Nick Telford <
> >>>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> Hi Bruno,
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> I've updated the KIP based on our conversation. The
> >>>> only
> >>>>>> things
> >>>>>>>>>>>>>>>> I've not
> >>>>>>>>>>>>>>>>>>> yet done are:
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> 1. Using transactions under ALOS and EOS.
> >>>>>>>>>>>>>>>>>>> 2. Making IsolationLevel a query-time constraint,
> >>>> rather
> >>>>>> than
> >>>>>>>>>>>>>>>> linking it
> >>>>>>>>>>>>>>>>>>> to the processing.guarantee.
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> There's a wrinkle that makes this a challenge:
> >>>> Interactive
> >>>>>>>>>> Queries
> >>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>> open an Iterator, when using transactions and
> >>>>>> READ_UNCOMMITTED.
> >>>>>>>>>>>>>>>>>>> The problem is that under READ_UNCOMMITTED, queries
> >>>> need
> >>>>>> to be
> >>>>>>>>>>>> able
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>> read records from the currently uncommitted
> >> transaction
> >>>>>> buffer
> >>>>>>>>>>>>>>>>>>> (WriteBatch). This includes for Iterators, which
> >> should
> >>>>>> iterate
> >>>>>>>>>>>>>>>> both the
> >>>>>>>>>>>>>>>>>>> transaction buffer and underlying database (using
> >>>>>>>>>>>>>>>>>>> WriteBatch#iteratorWithBase()).
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> The issue is that when the StreamThread commits, it
> >>>> writes
> >>>>>> the
> >>>>>>>>>>>>>>>> current
> >>>>>>>>>>>>>>>>>>> WriteBatch to RocksDB *and then clears the
> >> WriteBatch*.
> >>>>>>>> Clearing
> >>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>> WriteBatch while an Interactive Query holds an open
> >>>>>> Iterator on
> >>>>>>>>>> it
> >>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>>> invalidate the Iterator. Worse, it turns out that
> >>>> Iterators
> >>>>>>>> over
> >>>>>>>>>> a
> >>>>>>>>>>>>>>>>>>> WriteBatch become invalidated not just when the
> >>>> WriteBatch
> >>>>>> is
> >>>>>>>>>>>>>>>> cleared,
> >>>>>>>>>>>>>>>>>> but
> >>>>>>>>>>>>>>>>>>> also when the Iterators' current key receives a new
> >>>> write.
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> Now that I'm writing this, I remember that this is
> >> the
> >>>>>> major
> >>>>>>>>>>>> reason
> >>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>> I
> >>>>>>>>>>>>>>>>>>> switched the original design from having a
> >> query-time
> >>>>>>>>>>>>>>>> IsolationLevel to
> >>>>>>>>>>>>>>>>>>> having the IsolationLevel linked to the
> >>>> transactionality
> >>>>>> of the
> >>>>>>>>>>>>>>>> stores
> >>>>>>>>>>>>>>>>>>> themselves.
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> It *might* be possible to resolve this, by having a
> >>>>>> "chain" of
> >>>>>>>>>>>>>>>>>>> WriteBatches, with the StreamThread switching to a
> >> new
> >>>>>>>> WriteBatch
> >>>>>>>>>>>>>>>>>> whenever
> >>>>>>>>>>>>>>>>>>> a new Interactive Query attempts to read from the
> >>>>>> database, but
> >>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>> could
> >>>>>>>>>>>>>>>>>>> cause some performance problems/memory pressure
> >> when
> >>>>>> subjected
> >>>>>>>> to
> >>>>>>>>>>>> a
> >>>>>>>>>>>>>>>> high
> >>>>>>>>>>>>>>>>>>> Interactive Query load. It would also reduce the
> >>>>>> efficiency of
> >>>>>>>>>>>>>>>>>> WriteBatches
> >>>>>>>>>>>>>>>>>>> on-commit, as we'd have to write N WriteBatches,
> >> where
> >>>> N
> >>>>>> is the
> >>>>>>>>>>>>>>>> number of
> >>>>>>>>>>>>>>>>>>> Interactive Queries since the last commit.
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> I realise this is getting into the weeds of the
> >>>>>> implementation,
> >>>>>>>>>>>> and
> >>>>>>>>>>>>>>>> you'd
> >>>>>>>>>>>>>>>>>>> rather we focus on the API for now, but I think
> >> it's
> >>>>>> important
> >>>>>>>> to
> >>>>>>>>>>>>>>>>>> consider
> >>>>>>>>>>>>>>>>>>> how to implement the desired API, in case we come
> >> up
> >>>> with
> >>>>>> an
> >>>>>>>> API
> >>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>> cannot be implemented efficiently, or even at all!
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> Thoughts?
> >>>>>>>>>>>>>>>>>>> --
> >>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>> On Wed, 13 Sept 2023 at 13:03, Bruno Cadonna <
> >>>>>>>> cado...@apache.org
> >>>>>>>>>>>
> >>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> Hi Nick,
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> 6.
> >>>>>>>>>>>>>>>>>>>> Of course, you are right! My bad!
> >>>>>>>>>>>>>>>>>>>> Wiping out the state in the downgrading case is
> >> fine.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> 3a.
> >>>>>>>>>>>>>>>>>>>> Focus on the public facing changes for the KIP. We
> >>>> will
> >>>>>> manage
> >>>>>>>>>> to
> >>>>>>>>>>>>>>>> get
> >>>>>>>>>>>>>>>>>>>> the internals right. Regarding state stores that
> >> do
> >>>> not
> >>>>>>>> support
> >>>>>>>>>>>>>>>>>>>> READ_COMMITTED, they should throw an error stating
> >>>> that
> >>>>>> they
> >>>>>>>> do
> >>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>> support READ_COMMITTED. No need to adapt all state
> >>>> stores
> >>>>>>>>>>>>>>>> immediately.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> 3b.
> >>>>>>>>>>>>>>>>>>>> I am in favor of using transactions also for ALOS.
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>>>>> Bruno
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> On 9/13/23 11:57 AM, Nick Telford wrote:
> >>>>>>>>>>>>>>>>>>>>> Hi Bruno,
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> Thanks for getting back to me!
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> 2.
> >>>>>>>>>>>>>>>>>>>>> The fact that implementations can always track
> >>>> estimated
> >>>>>>>> memory
> >>>>>>>>>>>>>>>> usage
> >>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>> the wrapper is a good point. I can remove -1 as
> >> an
> >>>>>> option,
> >>>>>>>> and
> >>>>>>>>>>>>>>>> I'll
> >>>>>>>>>>>>>>>>>>>> clarify
> >>>>>>>>>>>>>>>>>>>>> the JavaDoc that 0 is not just for
> >> non-transactional
> >>>>>> stores,
> >>>>>>>>>>>>>>>> which is
> >>>>>>>>>>>>>>>>>>>>> currently misleading.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> 6.
> >>>>>>>>>>>>>>>>>>>>> The problem with catching the exception in the
> >>>> downgrade
> >>>>>>>>>> process
> >>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>> would require new code in the Kafka version being
> >>>>>> downgraded
> >>>>>>>>>> to.
> >>>>>>>>>>>>>>>> Since
> >>>>>>>>>>>>>>>>>>>>> users could conceivably downgrade to almost *any*
> >>>> older
> >>>>>>>> version
> >>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>> Kafka
> >>>>>>>>>>>>>>>>>>>>> Streams, I'm not sure how we could add that code?
> >>>>>>>>>>>>>>>>>>>>> The only way I can think of doing it would be to
> >>>> provide
> >>>>>> a
> >>>>>>>>>>>>>>>> dedicated
> >>>>>>>>>>>>>>>>>>>>> downgrade tool, that goes through every local
> >> store
> >>>> and
> >>>>>>>> removes
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> offsets column families. But that seems like an
> >>>>>> unnecessary
> >>>>>>>>>>>>>>>> amount of
> >>>>>>>>>>>>>>>>>>>> extra
> >>>>>>>>>>>>>>>>>>>>> code to maintain just to handle a somewhat niche
> >>>>>> situation,
> >>>>>>>>>> when
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> alternative (automatically wipe and restore
> >> stores)
> >>>>>> should be
> >>>>>>>>>>>>>>>>>>>> acceptable.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> 1, 4, 5: Agreed. I'll make the changes you've
> >>>> requested.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> 3a.
> >>>>>>>>>>>>>>>>>>>>> I agree that IsolationLevel makes more sense at
> >>>>>> query-time,
> >>>>>>>> and
> >>>>>>>>>>>> I
> >>>>>>>>>>>>>>>>>>>> actually
> >>>>>>>>>>>>>>>>>>>>> initially attempted to place the IsolationLevel
> >> at
> >>>>>>>> query-time,
> >>>>>>>>>>>>>>>> but I
> >>>>>>>>>>>>>>>>>> ran
> >>>>>>>>>>>>>>>>>>>>> into some problems:
> >>>>>>>>>>>>>>>>>>>>> - The key issue is that, under ALOS we're not
> >> staging
> >>>>>> writes
> >>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>> transactions, so can't perform writes at the
> >>>>>> READ_COMMITTED
> >>>>>>>>>>>>>>>> isolation
> >>>>>>>>>>>>>>>>>>>>> level. However, this may be addressed if we
> >> decide to
> >>>>>>>> *always*
> >>>>>>>>>>>>>>>> use
> >>>>>>>>>>>>>>>>>>>>> transactions as discussed under 3b.
> >>>>>>>>>>>>>>>>>>>>> - IQv1 and IQv2 have quite different
> >>>> implementations. I
> >>>>>>>>>> remember
> >>>>>>>>>>>>>>>>>> having
> >>>>>>>>>>>>>>>>>>>>> some difficulty understanding the IQv1 internals,
> >>>> which
> >>>>>> made
> >>>>>>>> it
> >>>>>>>>>>>>>>>>>>>> difficult
> >>>>>>>>>>>>>>>>>>>>> to determine what needed to be changed. However,
> >> I
> >>>>>> *think*
> >>>>>>>> this
> >>>>>>>>>>>>>>>> can be
> >>>>>>>>>>>>>>>>>>>>> addressed for both implementations by wrapping
> >> the
> >>>>>>>> RocksDBStore
> >>>>>>>>>>>>>>>> in an
> >>>>>>>>>>>>>>>>>>>>> IsolationLevel-dependent wrapper, that overrides
> >> read
> >>>>>> methods
> >>>>>>>>>>>>>>>> (get,
> >>>>>>>>>>>>>>>>>>>> etc.)
> >>>>>>>>>>>>>>>>>>>>> to either read directly from the database or
> >> from the
> >>>>>> ongoing
> >>>>>>>>>>>>>>>>>>>> transaction.
> >>>>>>>>>>>>>>>>>>>>> But IQv1 might still be difficult.
> >>>>>>>>>>>>>>>>>>>>> - If IsolationLevel becomes a query constraint,
> >> then
> >>>> all
> >>>>>>>> other
> >>>>>>>>>>>>>>>>>>>> StateStores
> >>>>>>>>>>>>>>>>>>>>> will need to respect it, including the in-memory
> >>>> stores.
> >>>>>> This
> >>>>>>>>>>>>>>>> would
> >>>>>>>>>>>>>>>>>>>> require
> >>>>>>>>>>>>>>>>>>>>> us to adapt in-memory stores to stage their
> >> writes so
> >>>>>> they
> >>>>>>>> can
> >>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>> isolated
> >>>>>>>>>>>>>>>>>>>>> from READ_COMMITTTED queries. It would also
> >> become an
> >>>>>>>> important
> >>>>>>>>>>>>>>>>>>>>> consideration for third-party stores on upgrade,
> >> as
> >>>>>> without
> >>>>>>>>>>>>>>>> changes,
> >>>>>>>>>>>>>>>>>>>> they
> >>>>>>>>>>>>>>>>>>>>> would not support READ_COMMITTED queries
> >> correctly.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> Ultimately, I may need some help making the
> >> necessary
> >>>>>> change
> >>>>>>>> to
> >>>>>>>>>>>>>>>> IQv1
> >>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>> support this, but I don't think it's
> >> fundamentally
> >>>>>>>> impossible,
> >>>>>>>>>>>>>>>> if we
> >>>>>>>>>>>>>>>>>>>> want
> >>>>>>>>>>>>>>>>>>>>> to pursue this route.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> 3b.
> >>>>>>>>>>>>>>>>>>>>> The main reason I chose to keep ALOS
> >> un-transactional
> >>>>>> was to
> >>>>>>>>>>>>>>>> minimize
> >>>>>>>>>>>>>>>>>>>>> behavioural change for most users (I believe most
> >>>> Streams
> >>>>>>>> users
> >>>>>>>>>>>>>>>> use
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> default configuration, which is ALOS). That said,
> >>>> it's
> >>>>>> clear
> >>>>>>>>>>>>>>>> that if
> >>>>>>>>>>>>>>>>>>>> ALOS
> >>>>>>>>>>>>>>>>>>>>> also used transactional stores, the only change
> >> in
> >>>>>> behaviour
> >>>>>>>>>>>>>>>> would be
> >>>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>> it would become *more correct*, which could be
> >>>>>> considered a
> >>>>>>>>>> "bug
> >>>>>>>>>>>>>>>> fix"
> >>>>>>>>>>>>>>>>>> by
> >>>>>>>>>>>>>>>>>>>>> users, rather than a change they need to handle.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> I believe that performance using transactions
> >> (aka.
> >>>>>> RocksDB
> >>>>>>>>>>>>>>>>>>>> WriteBatches)
> >>>>>>>>>>>>>>>>>>>>> should actually be *better* than the un-batched
> >>>>>> write-path
> >>>>>>>> that
> >>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>> currently used[1]. The only "performance"
> >>>> consideration
> >>>>>> will
> >>>>>>>> be
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>> increased memory usage that transactions require.
> >>>> Given
> >>>>>> the
> >>>>>>>>>>>>>>>>>> mitigations
> >>>>>>>>>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>>>>> this memory that we have in place, I would expect
> >>>> that
> >>>>>> this
> >>>>>>>> is
> >>>>>>>>>>>>>>>> not a
> >>>>>>>>>>>>>>>>>>>>> problem for most users.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> If we're happy to do so, we can make ALOS also
> >> use
> >>>>>>>>>> transactions.
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> Link 1:
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>
> >>>>
> >> https://github.com/adamretter/rocksjava-write-methods-benchmark#results
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>> On Wed, 13 Sept 2023 at 09:41, Bruno Cadonna <
> >>>>>>>>>>>> cado...@apache.org
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> Hi Nick,
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> Thanks for the updates and sorry for the delay
> >> on my
> >>>>>> side!
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 1.
> >>>>>>>>>>>>>>>>>>>>>> Making the default implementation for flush() a
> >>>> no-op
> >>>>>> sounds
> >>>>>>>>>>>>>>>> good to
> >>>>>>>>>>>>>>>>>>>> me.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 2.
> >>>>>>>>>>>>>>>>>>>>>> I think what was bugging me here is that a
> >>>> third-party
> >>>>>> state
> >>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>> needs
> >>>>>>>>>>>>>>>>>>>>>> to implement the state store interface. That
> >> means
> >>>> they
> >>>>>> need
> >>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>> implement a wrapper around the actual state
> >> store
> >>>> as we
> >>>>>> do
> >>>>>>>> for
> >>>>>>>>>>>>>>>>>> RocksDB
> >>>>>>>>>>>>>>>>>>>>>> with RocksDBStore. So, a third-party state
> >> store can
> >>>>>> always
> >>>>>>>>>>>>>>>> estimate
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> uncommitted bytes, if it wants, because the
> >> wrapper
> >>>> can
> >>>>>>>> record
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>> added
> >>>>>>>>>>>>>>>>>>>>>> bytes.
> >>>>>>>>>>>>>>>>>>>>>> One case I can think of where returning -1 makes
> >>>> sense
> >>>>>> is
> >>>>>>>> when
> >>>>>>>>>>>>>>>>>> Streams
> >>>>>>>>>>>>>>>>>>>>>> does not need to estimate the size of the write
> >>>> batch
> >>>>>> and
> >>>>>>>>>>>>>>>> trigger
> >>>>>>>>>>>>>>>>>>>>>> extraordinary commits, because the third-party
> >> state
> >>>>>> store
> >>>>>>>>>>>>>>>> takes care
> >>>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>>> memory. But in that case the method could also
> >> just
> >>>>>> return
> >>>>>>>> 0.
> >>>>>>>>>>>>>>>> Even
> >>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>> case would be better solved with a method that
> >>>> returns
> >>>>>>>> whether
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>>>>>> store manages itself the memory used for
> >> uncommitted
> >>>>>> bytes
> >>>>>>>> or
> >>>>>>>>>>>>>>>> not.
> >>>>>>>>>>>>>>>>>>>>>> Said that, I am fine with keeping the -1 return
> >>>> value,
> >>>>>> I was
> >>>>>>>>>>>>>>>> just
> >>>>>>>>>>>>>>>>>>>>>> wondering when and if it will be used.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> Regarding returning 0 for transactional state
> >> stores
> >>>>>> when
> >>>>>>>> the
> >>>>>>>>>>>>>>>> batch
> >>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>> empty, I was just wondering because you
> >> explicitly
> >>>>>> stated
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> "or {@code 0} if this StateStore does not
> >> support
> >>>>>>>>>>>> transactions."
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> So it seemed to me returning 0 could only
> >> happen for
> >>>>>>>>>>>>>>>>>> non-transactional
> >>>>>>>>>>>>>>>>>>>>>> state stores.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 3.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> a) What do you think if we move the isolation
> >> level
> >>>> to
> >>>>>> IQ
> >>>>>>>> (v1
> >>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>> v2)?
> >>>>>>>>>>>>>>>>>>>>>> In the end this is the only component that
> >> really
> >>>> needs
> >>>>>> to
> >>>>>>>>>>>>>>>> specify
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> isolation level. It is similar to the Kafka
> >> consumer
> >>>>>> that
> >>>>>>>> can
> >>>>>>>>>>>>>>>> choose
> >>>>>>>>>>>>>>>>>>>>>> with what isolation level to read the input
> >> topic.
> >>>>>>>>>>>>>>>>>>>>>> For IQv1 the isolation level should go into
> >>>>>>>>>>>>>>>> StoreQueryParameters. For
> >>>>>>>>>>>>>>>>>>>>>> IQv2, I would add it to the Query interface.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> b) Point a) raises the question what should
> >> happen
> >>>>>> during
> >>>>>>>>>>>>>>>>>> at-least-once
> >>>>>>>>>>>>>>>>>>>>>> processing when the state store does not use
> >>>>>> transactions?
> >>>>>>>>>> John
> >>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> past proposed to also use transactions on state
> >>>> stores
> >>>>>> for
> >>>>>>>>>>>>>>>>>>>>>> at-least-once. I like that idea, because it
> >> avoids
> >>>>>>>> aggregating
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>> same
> >>>>>>>>>>>>>>>>>>>>>> records over and over again in the case of a
> >>>> failure. We
> >>>>>>>> had a
> >>>>>>>>>>>>>>>> case
> >>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>> the past where a Streams applications in
> >>>> at-least-once
> >>>>>> mode
> >>>>>>>>>> was
> >>>>>>>>>>>>>>>>>> failing
> >>>>>>>>>>>>>>>>>>>>>> continuously for some reasons I do not remember
> >>>> before
> >>>>>>>>>>>>>>>> committing the
> >>>>>>>>>>>>>>>>>>>>>> offsets. After each failover, the app aggregated
> >>>> again
> >>>>>> and
> >>>>>>>>>>>>>>>> again the
> >>>>>>>>>>>>>>>>>>>>>> same records. Of course the aggregate increased
> >> to
> >>>> very
> >>>>>>>> wrong
> >>>>>>>>>>>>>>>> values
> >>>>>>>>>>>>>>>>>>>>>> just because of the failover. With transactions
> >> on
> >>>> the
> >>>>>> state
> >>>>>>>>>>>>>>>> stores
> >>>>>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>>>>>> could have avoided this. The app would have
> >> output
> >>>> the
> >>>>>> same
> >>>>>>>>>>>>>>>> aggregate
> >>>>>>>>>>>>>>>>>>>>>> multiple times (i.e., after each failover) but
> >> at
> >>>> least
> >>>>>> the
> >>>>>>>>>>>>>>>> value of
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> aggregate would not depend on the number of
> >>>> failovers.
> >>>>>>>>>>>>>>>> Outputting the
> >>>>>>>>>>>>>>>>>>>>>> same aggregate multiple times would be incorrect
> >>>> under
> >>>>>>>>>>>>>>>> exactly-once
> >>>>>>>>>>>>>>>>>> but
> >>>>>>>>>>>>>>>>>>>>>> it is OK for at-least-once.
> >>>>>>>>>>>>>>>>>>>>>> If it makes sense to add a config to turn on
> >> and off
> >>>>>>>>>>>>>>>> transactions on
> >>>>>>>>>>>>>>>>>>>>>> state stores under at-least-once or just use
> >>>>>> transactions in
> >>>>>>>>>>>>>>>> any case
> >>>>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>> a question we should also discuss in this KIP.
> >> It
> >>>>>> depends a
> >>>>>>>>>> bit
> >>>>>>>>>>>>>>>> on
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> performance trade-off. Maybe to be safe, I would
> >>>> add a
> >>>>>>>> config.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 4.
> >>>>>>>>>>>>>>>>>>>>>> Your points are all valid. I tend to say to
> >> keep the
> >>>>>> metrics
> >>>>>>>>>>>>>>>> around
> >>>>>>>>>>>>>>>>>>>>>> flush() until we remove flush() completely from
> >> the
> >>>>>>>> interface.
> >>>>>>>>>>>>>>>> Calls
> >>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>> flush() might still exist since existing
> >> processors
> >>>>>> might
> >>>>>>>>>> still
> >>>>>>>>>>>>>>>> call
> >>>>>>>>>>>>>>>>>>>>>> flush() explicitly as you mentioned in 1). For
> >>>> sure, we
> >>>>>> need
> >>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>> document
> >>>>>>>>>>>>>>>>>>>>>> how the metrics change due to the transactions
> >> in
> >>>> the
> >>>>>>>> upgrade
> >>>>>>>>>>>>>>>> notes.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 5.
> >>>>>>>>>>>>>>>>>>>>>> I see. Then you should describe how the
> >> .position
> >>>> files
> >>>>>> are
> >>>>>>>>>>>>>>>> handled
> >>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>> a dedicated section of the KIP or incorporate
> >> the
> >>>>>>>> description
> >>>>>>>>>>>>>>>> in the
> >>>>>>>>>>>>>>>>>>>>>> "Atomic Checkpointing" section instead of only
> >>>>>> mentioning it
> >>>>>>>>>> in
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> "Compatibility, Deprecation, and Migration
> >> Plan".
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> 6.
> >>>>>>>>>>>>>>>>>>>>>> Describing upgrading and downgrading in the KIP
> >> is a
> >>>>>> good
> >>>>>>>>>> idea.
> >>>>>>>>>>>>>>>>>>>>>> Regarding downgrading, I think you could also
> >> catch
> >>>> the
> >>>>>>>>>>>>>>>> exception and
> >>>>>>>>>>>>>>>>>>>> do
> >>>>>>>>>>>>>>>>>>>>>> what is needed to downgrade, e.g., drop the
> >> column
> >>>>>> family.
> >>>>>>>> See
> >>>>>>>>>>>>>>>> here
> >>>>>>>>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>>>>>> an example:
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>
> >>>>>>
> >>>>
> >> https://github.com/apache/kafka/blob/63fee01366e6ce98b9dfafd279a45d40b80e282d/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedStore.java#L75
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> It is a bit brittle, but it works.
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>>>>>>> Bruno
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>> On 8/24/23 12:18 PM, Nick Telford wrote:
> >>>>>>>>>>>>>>>>>>>>>>> Hi Bruno,
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Thanks for taking the time to review the KIP.
> >> I'm
> >>>> back
> >>>>>> from
> >>>>>>>>>>>>>>>> leave
> >>>>>>>>>>>>>>>>>> now
> >>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>> intend to move this forwards as quickly as I
> >> can.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Addressing your points:
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 1.
> >>>>>>>>>>>>>>>>>>>>>>> Because flush() is part of the StateStore API,
> >> it's
> >>>>>> exposed
> >>>>>>>>>> to
> >>>>>>>>>>>>>>>>>> custom
> >>>>>>>>>>>>>>>>>>>>>>> Processors, which might be making calls to
> >> flush().
> >>>>>> This
> >>>>>>>> was
> >>>>>>>>>>>>>>>>>> actually
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> case in a few integration tests.
> >>>>>>>>>>>>>>>>>>>>>>> To maintain as much compatibility as possible,
> >> I'd
> >>>>>> prefer
> >>>>>>>> not
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>> make
> >>>>>>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>>> an UnsupportedOperationException, as it will
> >> cause
> >>>>>>>> previously
> >>>>>>>>>>>>>>>>>> working
> >>>>>>>>>>>>>>>>>>>>>>> Processors to start throwing exceptions at
> >> runtime.
> >>>>>>>>>>>>>>>>>>>>>>> I agree that it doesn't make sense for it to
> >> proxy
> >>>>>>>> commit(),
> >>>>>>>>>>>>>>>> though,
> >>>>>>>>>>>>>>>>>>>> as
> >>>>>>>>>>>>>>>>>>>>>>> that would cause it to violate the "StateStores
> >>>> commit
> >>>>>> only
> >>>>>>>>>>>>>>>> when the
> >>>>>>>>>>>>>>>>>>>> Task
> >>>>>>>>>>>>>>>>>>>>>>> commits" rule.
> >>>>>>>>>>>>>>>>>>>>>>> Instead, I think we should make this a no-op.
> >> That
> >>>> way,
> >>>>>>>>>>>>>>>> existing
> >>>>>>>>>>>>>>>>>> user
> >>>>>>>>>>>>>>>>>>>>>>> Processors will continue to work as-before,
> >> without
> >>>>>>>> violation
> >>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>> consistency that would be caused by premature
> >>>>>> flush/commit
> >>>>>>>> of
> >>>>>>>>>>>>>>>>>>>> StateStore
> >>>>>>>>>>>>>>>>>>>>>>> data to disk.
> >>>>>>>>>>>>>>>>>>>>>>> What do you think?
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 2.
> >>>>>>>>>>>>>>>>>>>>>>> As stated in the JavaDoc, when a StateStore
> >>>>>> implementation
> >>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>>> transactional, but is unable to estimate the
> >>>>>> uncommitted
> >>>>>>>>>>>> memory
> >>>>>>>>>>>>>>>>>> usage,
> >>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> method will return -1.
> >>>>>>>>>>>>>>>>>>>>>>> The intention here is to permit third-party
> >>>>>> implementations
> >>>>>>>>>>>>>>>> that may
> >>>>>>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>> able to estimate memory usage.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Yes, it will be 0 when nothing has been
> >> written to
> >>>> the
> >>>>>>>> store
> >>>>>>>>>>>>>>>> yet. I
> >>>>>>>>>>>>>>>>>>>>>> thought
> >>>>>>>>>>>>>>>>>>>>>>> that was implied by "This method will return an
> >>>>>>>> approximation
> >>>>>>>>>>>>>>>> of the
> >>>>>>>>>>>>>>>>>>>>>> memory
> >>>>>>>>>>>>>>>>>>>>>>> would be freed by the next call to {@link
> >>>>>> #commit(Map)}"
> >>>>>>>> and
> >>>>>>>>>>>>>>>>>> "@return
> >>>>>>>>>>>>>>>>>>>> The
> >>>>>>>>>>>>>>>>>>>>>>> approximate size of all records awaiting {@link
> >>>>>>>>>>>> #commit(Map)}",
> >>>>>>>>>>>>>>>>>>>> however,
> >>>>>>>>>>>>>>>>>>>>>> I
> >>>>>>>>>>>>>>>>>>>>>>> can add it explicitly to the JavaDoc if you
> >> think
> >>>> this
> >>>>>> is
> >>>>>>>>>>>>>>>> unclear?
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 3.
> >>>>>>>>>>>>>>>>>>>>>>> I realise this is probably the most contentious
> >>>> point
> >>>>>> in my
> >>>>>>>>>>>>>>>> design,
> >>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>> I'm
> >>>>>>>>>>>>>>>>>>>>>>> open to changing it if I'm unable to convince
> >> you
> >>>> of
> >>>>>> the
> >>>>>>>>>>>>>>>> benefits.
> >>>>>>>>>>>>>>>>>>>>>>> Nevertheless, here's my argument:
> >>>>>>>>>>>>>>>>>>>>>>> The Interactive Query (IQ) API(s) are directly
> >>>> provided
> >>>>>>>>>>>>>>>> StateStores
> >>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>> query, and it may be important for users to
> >>>>>>>> programmatically
> >>>>>>>>>>>>>>>> know
> >>>>>>>>>>>>>>>>>>>> which
> >>>>>>>>>>>>>>>>>>>>>>> mode the StateStore is operating under. If we
> >>>> simply
> >>>>>>>> provide
> >>>>>>>>>>>> an
> >>>>>>>>>>>>>>>>>>>>>>> "eosEnabled" boolean (as used throughout the
> >>>> internal
> >>>>>>>> streams
> >>>>>>>>>>>>>>>>>>>> engine), or
> >>>>>>>>>>>>>>>>>>>>>>> similar, then users will need to understand the
> >>>>>> operation
> >>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>> consequences
> >>>>>>>>>>>>>>>>>>>>>>> of each available processing mode and how it
> >>>> pertains
> >>>>>> to
> >>>>>>>>>> their
> >>>>>>>>>>>>>>>>>>>>>> StateStore.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Interactive Query users aren't the only people
> >> that
> >>>>>> care
> >>>>>>>>>> about
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> processing.mode/IsolationLevel of a StateStore:
> >>>>>>>> implementers
> >>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>> custom
> >>>>>>>>>>>>>>>>>>>>>>> StateStores also need to understand the
> >> behaviour
> >>>>>> expected
> >>>>>>>> of
> >>>>>>>>>>>>>>>> their
> >>>>>>>>>>>>>>>>>>>>>>> implementation. KIP-892 introduces some
> >> assumptions
> >>>>>> into
> >>>>>>>> the
> >>>>>>>>>>>>>>>> Streams
> >>>>>>>>>>>>>>>>>>>>>> Engine
> >>>>>>>>>>>>>>>>>>>>>>> about how StateStores operate under each
> >> processing
> >>>>>> mode,
> >>>>>>>> and
> >>>>>>>>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>>>>> important that custom implementations adhere to
> >>>> those
> >>>>>>>>>>>>>>>> assumptions in
> >>>>>>>>>>>>>>>>>>>>>> order
> >>>>>>>>>>>>>>>>>>>>>>> to maintain the consistency guarantees.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> IsolationLevels provide a high-level contract
> >> on
> >>>> the
> >>>>>>>>>> behaviour
> >>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> StateStore: a user knows that under
> >> READ_COMMITTED,
> >>>>>> they
> >>>>>>>> will
> >>>>>>>>>>>>>>>> see
> >>>>>>>>>>>>>>>>>>>> writes
> >>>>>>>>>>>>>>>>>>>>>>> only after the Task has committed, and under
> >>>>>>>> READ_UNCOMMITTED
> >>>>>>>>>>>>>>>> they
> >>>>>>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>>>>>> see
> >>>>>>>>>>>>>>>>>>>>>>> writes immediately. No understanding of the
> >>>> details of
> >>>>>> each
> >>>>>>>>>>>>>>>>>>>>>> processing.mode
> >>>>>>>>>>>>>>>>>>>>>>> is required, either for IQ users or StateStore
> >>>>>>>> implementers.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> An argument can be made that these contractual
> >>>>>> guarantees
> >>>>>>>> can
> >>>>>>>>>>>>>>>> simply
> >>>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>> documented for the processing.mode (i.e. that
> >>>>>> exactly-once
> >>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>> exactly-once-v2 behave like READ_COMMITTED and
> >>>>>>>> at-least-once
> >>>>>>>>>>>>>>>> behaves
> >>>>>>>>>>>>>>>>>>>> like
> >>>>>>>>>>>>>>>>>>>>>>> READ_UNCOMMITTED), but there are several small
> >>>> issues
> >>>>>> with
> >>>>>>>>>>>>>>>> this I'd
> >>>>>>>>>>>>>>>>>>>>>> prefer
> >>>>>>>>>>>>>>>>>>>>>>> to avoid:
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> - Where would we document these
> >> contracts,
> >>>> in
> >>>>>> a way
> >>>>>>>>>> that
> >>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>> difficult
> >>>>>>>>>>>>>>>>>>>>>>> for users/implementers to miss/ignore?
> >>>>>>>>>>>>>>>>>>>>>>> - It's not clear to users that the
> >>>> processing
> >>>>>> mode
> >>>>>>>> is
> >>>>>>>>>>>>>>>>>>>> communicating
> >>>>>>>>>>>>>>>>>>>>>>> an expectation of read isolation,
> >> unless
> >>>> they
> >>>>>> read
> >>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> documentation. Users
> >>>>>>>>>>>>>>>>>>>>>>> rarely consult documentation unless
> >> they
> >>>> feel
> >>>>>> they
> >>>>>>>>>> need
> >>>>>>>>>>>>>>>> to, so
> >>>>>>>>>>>>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>>>> likely
> >>>>>>>>>>>>>>>>>>>>>>> this detail would get missed by many
> >> users.
> >>>>>>>>>>>>>>>>>>>>>>> - It tightly couples processing modes
> >> to
> >>>> read
> >>>>>>>>>> isolation.
> >>>>>>>>>>>>>>>> Adding
> >>>>>>>>>>>>>>>>>>>> new
> >>>>>>>>>>>>>>>>>>>>>>> processing modes, or changing the read
> >>>>>> isolation of
> >>>>>>>>>>>>>>>> existing
> >>>>>>>>>>>>>>>>>>>>>> processing
> >>>>>>>>>>>>>>>>>>>>>>> modes would be difficult/impossible.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Ultimately, the cost of introducing
> >>>> IsolationLevels is
> >>>>>>>> just a
> >>>>>>>>>>>>>>>> single
> >>>>>>>>>>>>>>>>>>>>>>> method, since we re-use the existing
> >> IsolationLevel
> >>>>>> enum
> >>>>>>>> from
> >>>>>>>>>>>>>>>> Kafka.
> >>>>>>>>>>>>>>>>>>>> This
> >>>>>>>>>>>>>>>>>>>>>>> gives us a clear place to document the
> >> contractual
> >>>>>>>> guarantees
> >>>>>>>>>>>>>>>>>> expected
> >>>>>>>>>>>>>>>>>>>>>>> of/provided by StateStores, that is accessible
> >>>> both by
> >>>>>> the
> >>>>>>>>>>>>>>>>>> StateStore
> >>>>>>>>>>>>>>>>>>>>>>> itself, and by IQ users.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> (Writing this I've just realised that the
> >>>> StateStore
> >>>>>> and IQ
> >>>>>>>>>>>>>>>> APIs
> >>>>>>>>>>>>>>>>>>>> actually
> >>>>>>>>>>>>>>>>>>>>>>> don't provide access to StateStoreContext that
> >> IQ
> >>>> users
> >>>>>>>> would
> >>>>>>>>>>>>>>>> have
> >>>>>>>>>>>>>>>>>>>> direct
> >>>>>>>>>>>>>>>>>>>>>>> access to... Perhaps StateStore should expose
> >>>>>>>>>> isolationLevel()
> >>>>>>>>>>>>>>>>>> itself
> >>>>>>>>>>>>>>>>>>>>>> too?)
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 4.
> >>>>>>>>>>>>>>>>>>>>>>> Yeah, I'm not comfortable renaming the metrics
> >>>> in-place
> >>>>>>>>>>>>>>>> either, as
> >>>>>>>>>>>>>>>>>>>> it's a
> >>>>>>>>>>>>>>>>>>>>>>> backwards incompatible change. My concern is
> >> that,
> >>>> if
> >>>>>> we
> >>>>>>>>>> leave
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> existing
> >>>>>>>>>>>>>>>>>>>>>>> "flush" metrics in place, they will be
> >> confusing to
> >>>>>> users.
> >>>>>>>>>>>>>>>> Right
> >>>>>>>>>>>>>>>>>> now,
> >>>>>>>>>>>>>>>>>>>>>>> "flush" metrics record explicit flushes to
> >> disk,
> >>>> but
> >>>>>> under
> >>>>>>>>>>>>>>>> KIP-892,
> >>>>>>>>>>>>>>>>>>>> even
> >>>>>>>>>>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>>>>>>> commit() will not explicitly flush data to
> >> disk -
> >>>>>> RocksDB
> >>>>>>>>>> will
> >>>>>>>>>>>>>>>>>> decide
> >>>>>>>>>>>>>>>>>>>> on
> >>>>>>>>>>>>>>>>>>>>>>> when to flush memtables to disk itself.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> If we keep the existing "flush" metrics, we'd
> >> have
> >>>> two
> >>>>>>>>>>>> options,
> >>>>>>>>>>>>>>>>>> which
> >>>>>>>>>>>>>>>>>>>>>> both
> >>>>>>>>>>>>>>>>>>>>>>> seem pretty bad to me:
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 1. Have them record calls to commit(),
> >>>> which
> >>>>>> would
> >>>>>>>> be
> >>>>>>>>>>>>>>>>>>>> misleading, as
> >>>>>>>>>>>>>>>>>>>>>>> data is no longer explicitly "flushed"
> >> to
> >>>> disk
> >>>>>> by
> >>>>>>>> this
> >>>>>>>>>>>>>>>> call.
> >>>>>>>>>>>>>>>>>>>>>>> 2. Have them record nothing at all,
> >> which
> >>>> is
> >>>>>>>>>> equivalent
> >>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>> removing
> >>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>> metrics, except that users will see the
> >>>> metric
> >>>>>>>> still
> >>>>>>>>>>>>>>>> exists and
> >>>>>>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>>>>> assume
> >>>>>>>>>>>>>>>>>>>>>>> that the metric is correct, and that
> >>>> there's a
> >>>>>>>> problem
> >>>>>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>> their
> >>>>>>>>>>>>>>>>>>>>>> system
> >>>>>>>>>>>>>>>>>>>>>>> when there isn't.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> I agree that removing them is also a bad
> >> solution,
> >>>> and
> >>>>>> I'd
> >>>>>>>>>>>>>>>> like some
> >>>>>>>>>>>>>>>>>>>>>>> guidance on the best path forward here.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 5.
> >>>>>>>>>>>>>>>>>>>>>>> Position files are updated on every write to a
> >>>>>> StateStore.
> >>>>>>>>>>>>>>>> Since our
> >>>>>>>>>>>>>>>>>>>>>> writes
> >>>>>>>>>>>>>>>>>>>>>>> are now buffered until commit(), we can't
> >> update
> >>>> the
> >>>>>>>> Position
> >>>>>>>>>>>>>>>> file
> >>>>>>>>>>>>>>>>>>>> until
> >>>>>>>>>>>>>>>>>>>>>>> commit() has been called, otherwise it would be
> >>>>>>>> inconsistent
> >>>>>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> data
> >>>>>>>>>>>>>>>>>>>>>>> in the event of a rollback. Consequently, we
> >> need
> >>>> to
> >>>>>> manage
> >>>>>>>>>>>>>>>> these
> >>>>>>>>>>>>>>>>>>>> offsets
> >>>>>>>>>>>>>>>>>>>>>>> the same way we manage the checkpoint offsets,
> >> and
> >>>>>> ensure
> >>>>>>>>>>>>>>>> they're
> >>>>>>>>>>>>>>>>>> only
> >>>>>>>>>>>>>>>>>>>>>>> written on commit().
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> 6.
> >>>>>>>>>>>>>>>>>>>>>>> Agreed, although I'm not exactly sure yet what
> >>>> tests to
> >>>>>>>>>> write.
> >>>>>>>>>>>>>>>> How
> >>>>>>>>>>>>>>>>>>>>>> explicit
> >>>>>>>>>>>>>>>>>>>>>>> do we need to be here in the KIP?
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> As for upgrade/downgrade: upgrade is designed
> >> to be
> >>>>>>>> seamless,
> >>>>>>>>>>>>>>>> and we
> >>>>>>>>>>>>>>>>>>>>>> should
> >>>>>>>>>>>>>>>>>>>>>>> definitely add some tests around that.
> >> Downgrade,
> >>>> it
> >>>>>>>>>>>>>>>> transpires,
> >>>>>>>>>>>>>>>>>> isn't
> >>>>>>>>>>>>>>>>>>>>>>> currently possible, as the extra column family
> >> for
> >>>>>> offset
> >>>>>>>>>>>>>>>> storage is
> >>>>>>>>>>>>>>>>>>>>>>> incompatible with the pre-KIP-892
> >> implementation:
> >>>> when
> >>>>>> you
> >>>>>>>>>>>>>>>> open a
> >>>>>>>>>>>>>>>>>>>> RocksDB
> >>>>>>>>>>>>>>>>>>>>>>> database, you must open all available column
> >>>> families
> >>>>>> or
> >>>>>>>>>>>>>>>> receive an
> >>>>>>>>>>>>>>>>>>>>>> error.
> >>>>>>>>>>>>>>>>>>>>>>> What currently happens on downgrade is that it
> >>>>>> attempts to
> >>>>>>>>>>>>>>>> open the
> >>>>>>>>>>>>>>>>>>>>>> store,
> >>>>>>>>>>>>>>>>>>>>>>> throws an error about the offsets column
> >> family not
> >>>>>> being
> >>>>>>>>>>>>>>>> opened,
> >>>>>>>>>>>>>>>>>>>> which
> >>>>>>>>>>>>>>>>>>>>>>> triggers a wipe and rebuild of the Task. Given
> >> that
> >>>>>>>>>> downgrades
> >>>>>>>>>>>>>>>>>> should
> >>>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>> uncommon, I think this is acceptable
> >> behaviour, as
> >>>> the
> >>>>>>>>>>>>>>>> end-state is
> >>>>>>>>>>>>>>>>>>>>>>> consistent, even if it results in an
> >> undesirable
> >>>> state
> >>>>>>>>>>>> restore.
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Should I document the upgrade/downgrade
> >> behaviour
> >>>>>>>> explicitly
> >>>>>>>>>>>>>>>> in the
> >>>>>>>>>>>>>>>>>>>> KIP?
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> --
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>> On Mon, 14 Aug 2023 at 22:31, Bruno Cadonna <
> >>>>>>>>>>>>>>>> cado...@apache.org>
> >>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> Hi Nick!
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> Thanks for the updates!
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> 1.
> >>>>>>>>>>>>>>>>>>>>>>>> Why does StateStore#flush() default to
> >>>>>>>>>>>>>>>>>>>>>>>> StateStore#commit(Collections.emptyMap())?
> >>>>>>>>>>>>>>>>>>>>>>>> Since calls to flush() will not exist anymore
> >>>> after
> >>>>>> this
> >>>>>>>> KIP
> >>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>>>> released, I would rather throw an unsupported
> >>>>>> operation
> >>>>>>>>>>>>>>>> exception
> >>>>>>>>>>>>>>>>>> by
> >>>>>>>>>>>>>>>>>>>>>>>> default.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> 2.
> >>>>>>>>>>>>>>>>>>>>>>>> When would a state store return -1 from
> >>>>>>>>>>>>>>>>>>>>>>>> StateStore#approximateNumUncommittedBytes()
> >> while
> >>>>>> being
> >>>>>>>>>>>>>>>>>>>> transactional?
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> Wouldn't
> >>>> StateStore#approximateNumUncommittedBytes()
> >>>>>> also
> >>>>>>>>>>>>>>>> return 0
> >>>>>>>>>>>>>>>>>> if
> >>>>>>>>>>>>>>>>>>>>>>>> the state store is transactional but nothing
> >> has
> >>>> been
> >>>>>>>>>> written
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>> state store yet?
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> 3.
> >>>>>>>>>>>>>>>>>>>>>>>> Sorry for bringing this up again. Does this
> >> KIP
> >>>> really
> >>>>>>>> need
> >>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>> introduce
> >>>>>>>>>>>>>>>>>>>>>>>> StateStoreContext#isolationLevel()?
> >>>> StateStoreContext
> >>>>>> has
> >>>>>>>>>>>>>>>> already
> >>>>>>>>>>>>>>>>>>>>>>>> appConfigs() which basically exposes the same
> >>>>>> information,
> >>>>>>>>>>>>>>>> i.e., if
> >>>>>>>>>>>>>>>>>>>> EOS
> >>>>>>>>>>>>>>>>>>>>>>>> is enabled or not.
> >>>>>>>>>>>>>>>>>>>>>>>> In one of your previous e-mails you wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> "My idea was to try to keep the StateStore
> >>>> interface
> >>>>>> as
> >>>>>>>>>>>>>>>> loosely
> >>>>>>>>>>>>>>>>>>>> coupled
> >>>>>>>>>>>>>>>>>>>>>>>> from the Streams engine as possible, to give
> >>>>>> implementers
> >>>>>>>>>>>> more
> >>>>>>>>>>>>>>>>>>>> freedom,
> >>>>>>>>>>>>>>>>>>>>>>>> and reduce the amount of internal knowledge
> >>>> required."
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> While I understand the intent, I doubt that it
> >>>>>> decreases
> >>>>>>>> the
> >>>>>>>>>>>>>>>>>>>> coupling of
> >>>>>>>>>>>>>>>>>>>>>>>> a StateStore interface and the Streams engine.
> >>>>>>>>>> READ_COMMITTED
> >>>>>>>>>>>>>>>> only
> >>>>>>>>>>>>>>>>>>>>>>>> applies to IQ but not to reads by processors.
> >>>> Thus,
> >>>>>>>>>>>>>>>> implementers
> >>>>>>>>>>>>>>>>>>>> need to
> >>>>>>>>>>>>>>>>>>>>>>>> understand how Streams accesses the state
> >> stores.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> I would like to hear what others think about
> >> this.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> 4.
> >>>>>>>>>>>>>>>>>>>>>>>> Great exposing new metrics for transactional
> >> state
> >>>>>> stores!
> >>>>>>>>>>>>>>>>>> However, I
> >>>>>>>>>>>>>>>>>>>>>>>> would prefer to add new metrics and deprecate
> >> (in
> >>>> the
> >>>>>>>> docs)
> >>>>>>>>>>>>>>>> the old
> >>>>>>>>>>>>>>>>>>>>>>>> ones. You can find examples of deprecated
> >> metrics
> >>>>>> here:
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>> https://kafka.apache.org/documentation/#selector_monitoring
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> 5.
> >>>>>>>>>>>>>>>>>>>>>>>> Why does the KIP mention position files? I do
> >> not
> >>>>>> think
> >>>>>>>> they
> >>>>>>>>>>>>>>>> are
> >>>>>>>>>>>>>>>>>>>> related
> >>>>>>>>>>>>>>>>>>>>>>>> to transactions or flushes.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> 6.
> >>>>>>>>>>>>>>>>>>>>>>>> I think we will also need to adapt/add
> >> integration
> >>>>>> tests
> >>>>>>>>>>>>>>>> besides
> >>>>>>>>>>>>>>>>>> unit
> >>>>>>>>>>>>>>>>>>>>>>>> tests. Additionally, we probably need
> >> integration
> >>>> or
> >>>>>>>> system
> >>>>>>>>>>>>>>>> tests
> >>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>> verify that upgrades and downgrades between
> >>>>>> transactional
> >>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>>> non-transactional state stores work as
> >> expected.
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>>>>>>>>> Bruno
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>> On 7/21/23 10:34 PM, Nick Telford wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>> One more thing: I noted John's suggestion in
> >> the
> >>>> KIP,
> >>>>>>>> under
> >>>>>>>>>>>>>>>>>>>> "Rejected
> >>>>>>>>>>>>>>>>>>>>>>>>> Alternatives". I still think it's an idea
> >> worth
> >>>>>> pursuing,
> >>>>>>>>>>>>>>>> but I
> >>>>>>>>>>>>>>>>>>>> believe
> >>>>>>>>>>>>>>>>>>>>>>>>> that it's out of the scope of this KIP,
> >> because
> >>>> it
> >>>>>>>> solves a
> >>>>>>>>>>>>>>>>>>>> different
> >>>>>>>>>>>>>>>>>>>>>> set
> >>>>>>>>>>>>>>>>>>>>>>>>> of problems to this KIP, and the scope of
> >> this
> >>>> one
> >>>>>> has
> >>>>>>>>>>>>>>>> already
> >>>>>>>>>>>>>>>>>> grown
> >>>>>>>>>>>>>>>>>>>>>>>> quite
> >>>>>>>>>>>>>>>>>>>>>>>>> large!
> >>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>> On Fri, 21 Jul 2023 at 21:33, Nick Telford <
> >>>>>>>>>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> I've updated the KIP (
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>
> >>>>>>
> >>>>
> >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-892%3A+Transactional+Semantics+for+StateStores
> >>>>>>>>>>>>>>>>>>>>>>>> )
> >>>>>>>>>>>>>>>>>>>>>>>>>> with the latest changes; mostly bringing
> >> back
> >>>>>> "Atomic
> >>>>>>>>>>>>>>>>>>>> Checkpointing"
> >>>>>>>>>>>>>>>>>>>>>>>> (for
> >>>>>>>>>>>>>>>>>>>>>>>>>> what feels like the 10th time!). I think
> >> the one
> >>>>>> thing
> >>>>>>>>>>>>>>>> missing is
> >>>>>>>>>>>>>>>>>>>> some
> >>>>>>>>>>>>>>>>>>>>>>>>>> changes to metrics (notably the store
> >> "flush"
> >>>>>> metrics
> >>>>>>>> will
> >>>>>>>>>>>>>>>> need
> >>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>>>>> renamed to "commit").
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> The reason I brought back Atomic
> >> Checkpointing
> >>>> was
> >>>>>> to
> >>>>>>>>>>>>>>>> decouple
> >>>>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>>> flush
> >>>>>>>>>>>>>>>>>>>>>>>>>> from store commit. This is important,
> >> because
> >>>> with
> >>>>>>>>>>>>>>>> Transactional
> >>>>>>>>>>>>>>>>>>>>>>>>>> StateStores, we now need to call "flush" on
> >>>> *every*
> >>>>>> Task
> >>>>>>>>>>>>>>>> commit,
> >>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>>>>>>> just when the StateStore is closing,
> >> otherwise
> >>>> our
> >>>>>>>>>>>>>>>> transaction
> >>>>>>>>>>>>>>>>>>>> buffer
> >>>>>>>>>>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>>>>>>>>>> never be written and persisted, instead
> >> growing
> >>>>>>>> unbounded!
> >>>>>>>>>>>> I
> >>>>>>>>>>>>>>>>>>>>>>>> experimented
> >>>>>>>>>>>>>>>>>>>>>>>>>> with some simple solutions, like forcing a
> >> store
> >>>>>> flush
> >>>>>>>>>>>>>>>> whenever
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>> transaction buffer was likely to exceed its
> >>>>>> configured
> >>>>>>>>>>>>>>>> size, but
> >>>>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>>>> was
> >>>>>>>>>>>>>>>>>>>>>>>>>> brittle: it prevented the transaction buffer
> >>>> from
> >>>>>> being
> >>>>>>>>>>>>>>>>>> configured
> >>>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>>>>> unbounded, and it still would have required
> >>>> explicit
> >>>>>>>>>>>>>>>> flushes of
> >>>>>>>>>>>>>>>>>>>>>> RocksDB,
> >>>>>>>>>>>>>>>>>>>>>>>>>> yielding sub-optimal performance and memory
> >>>>>> utilization.
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> I deemed Atomic Checkpointing to be the
> >> "right"
> >>>> way
> >>>>>> to
> >>>>>>>>>>>>>>>> resolve
> >>>>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>>>>>>>>>> problem. By ensuring that the changelog
> >> offsets
> >>>> that
> >>>>>>>>>>>>>>>> correspond
> >>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>> most
> >>>>>>>>>>>>>>>>>>>>>>>>>> recently written records are always
> >> atomically
> >>>>>> written
> >>>>>>>> to
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> StateStore
> >>>>>>>>>>>>>>>>>>>>>>>>>> (by writing them to the same transaction
> >>>> buffer),
> >>>>>> we can
> >>>>>>>>>>>>>>>> avoid
> >>>>>>>>>>>>>>>>>>>>>> forcibly
> >>>>>>>>>>>>>>>>>>>>>>>>>> flushing the RocksDB memtables to disk,
> >> letting
> >>>>>> RocksDB
> >>>>>>>>>>>>>>>> flush
> >>>>>>>>>>>>>>>>>> them
> >>>>>>>>>>>>>>>>>>>>>> only
> >>>>>>>>>>>>>>>>>>>>>>>>>> when necessary, without losing any of our
> >>>>>> consistency
> >>>>>>>>>>>>>>>> guarantees.
> >>>>>>>>>>>>>>>>>>>> See
> >>>>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>> updated KIP for more info.
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> I have fully implemented these changes,
> >>>> although I'm
> >>>>>>>> still
> >>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>>> entirely
> >>>>>>>>>>>>>>>>>>>>>>>>>> happy with the implementation for segmented
> >>>>>> StateStores,
> >>>>>>>>>> so
> >>>>>>>>>>>>>>>> I
> >>>>>>>>>>>>>>>>>> plan
> >>>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>>>> refactor that. Despite that, all tests
> >> pass. If
> >>>>>> you'd
> >>>>>>>> like
> >>>>>>>>>>>>>>>> to try
> >>>>>>>>>>>>>>>>>>>> out
> >>>>>>>>>>>>>>>>>>>>>> or
> >>>>>>>>>>>>>>>>>>>>>>>>>> review this highly experimental and
> >> incomplete
> >>>>>> branch,
> >>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>> available
> >>>>>>>>>>>>>>>>>>>>>>>> here:
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>> https://github.com/nicktelford/kafka/tree/KIP-892-3.5.0
> >>>>>>>> .
> >>>>>>>>>>>>>>>> Note:
> >>>>>>>>>>>>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>>>>>> built
> >>>>>>>>>>>>>>>>>>>>>>>>>> against Kafka 3.5.0 so that I had a stable
> >> base
> >>>> to
> >>>>>> build
> >>>>>>>>>>>>>>>> and test
> >>>>>>>>>>>>>>>>>>>> it
> >>>>>>>>>>>>>>>>>>>>>> on,
> >>>>>>>>>>>>>>>>>>>>>>>>>> and to enable easy apples-to-apples
> >> comparisons
> >>>> in a
> >>>>>>>> live
> >>>>>>>>>>>>>>>>>>>>>> environment. I
> >>>>>>>>>>>>>>>>>>>>>>>>>> plan to rebase it against trunk once it's
> >> nearer
> >>>>>>>>>> completion
> >>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>> has
> >>>>>>>>>>>>>>>>>>>>>> been
> >>>>>>>>>>>>>>>>>>>>>>>>>> proven on our main application.
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> I would really appreciate help in reviewing
> >> and
> >>>>>> testing:
> >>>>>>>>>>>>>>>>>>>>>>>>>> - Segmented (Versioned, Session and Window)
> >>>> stores
> >>>>>>>>>>>>>>>>>>>>>>>>>> - Global stores
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> As I do not currently use either of these,
> >> so my
> >>>>>> primary
> >>>>>>>>>>>>>>>> test
> >>>>>>>>>>>>>>>>>>>>>>>> environment
> >>>>>>>>>>>>>>>>>>>>>>>>>> doesn't test these areas.
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> I'm going on Parental Leave starting next
> >> week
> >>>> for
> >>>>>> a few
> >>>>>>>>>>>>>>>> weeks,
> >>>>>>>>>>>>>>>>>> so
> >>>>>>>>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>>>>>>>>>> not have time to move this forward until
> >> late
> >>>>>> August.
> >>>>>>>> That
> >>>>>>>>>>>>>>>> said,
> >>>>>>>>>>>>>>>>>>>> your
> >>>>>>>>>>>>>>>>>>>>>>>>>> feedback is welcome and appreciated, I just
> >>>> won't be
> >>>>>>>> able
> >>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>> respond
> >>>>>>>>>>>>>>>>>>>>>> as
> >>>>>>>>>>>>>>>>>>>>>>>>>> quickly as usual.
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>> On Mon, 3 Jul 2023 at 16:23, Nick Telford <
> >>>>>>>>>>>>>>>>>> nick.telf...@gmail.com>
> >>>>>>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Bruno
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> Yes, that's correct, although the impact
> >> on IQ
> >>>> is
> >>>>>> not
> >>>>>>>>>>>>>>>> something
> >>>>>>>>>>>>>>>>>> I
> >>>>>>>>>>>>>>>>>>>> had
> >>>>>>>>>>>>>>>>>>>>>>>>>>> considered.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> What about atomically updating the state
> >> store
> >>>>>> from the
> >>>>>>>>>>>>>>>>>>>> transaction
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> buffer every commit interval and writing
> >> the
> >>>>>>>> checkpoint
> >>>>>>>>>>>>>>>> (thus,
> >>>>>>>>>>>>>>>>>>>>>>>> flushing
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> the memtable) every configured amount of
> >> data
> >>>>>> and/or
> >>>>>>>>>>>>>>>> number of
> >>>>>>>>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> intervals?
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> I'm not quite sure I follow. Are you
> >> suggesting
> >>>>>> that we
> >>>>>>>>>>>>>>>> add an
> >>>>>>>>>>>>>>>>>>>>>>>> additional
> >>>>>>>>>>>>>>>>>>>>>>>>>>> config for the max number of commit
> >> intervals
> >>>>>> between
> >>>>>>>>>>>>>>>>>> checkpoints?
> >>>>>>>>>>>>>>>>>>>>>> That
> >>>>>>>>>>>>>>>>>>>>>>>>>>> way, we would checkpoint *either* when the
> >>>>>> transaction
> >>>>>>>>>>>>>>>> buffers
> >>>>>>>>>>>>>>>>>> are
> >>>>>>>>>>>>>>>>>>>>>>>> nearly
> >>>>>>>>>>>>>>>>>>>>>>>>>>> full, *OR* whenever a certain number of
> >> commit
> >>>>>>>> intervals
> >>>>>>>>>>>>>>>> have
> >>>>>>>>>>>>>>>>>>>>>> elapsed,
> >>>>>>>>>>>>>>>>>>>>>>>>>>> whichever comes first?
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> That certainly seems reasonable, although
> >> this
> >>>>>>>> re-ignites
> >>>>>>>>>>>>>>>> an
> >>>>>>>>>>>>>>>>>>>> earlier
> >>>>>>>>>>>>>>>>>>>>>>>>>>> debate about whether a config should be
> >>>> measured in
> >>>>>>>>>>>>>>>> "number of
> >>>>>>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>>>>>>>> intervals", instead of just an absolute
> >> time.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> FWIW, I realised that this issue is the
> >> reason
> >>>> I
> >>>>>> was
> >>>>>>>>>>>>>>>> pursuing
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>> Atomic
> >>>>>>>>>>>>>>>>>>>>>>>>>>> Checkpoints, as it de-couples memtable
> >> flush
> >>>> from
> >>>>>>>>>>>>>>>> checkpointing,
> >>>>>>>>>>>>>>>>>>>>>> which
> >>>>>>>>>>>>>>>>>>>>>>>>>>> enables us to just checkpoint on every
> >> commit
> >>>>>> without
> >>>>>>>> any
> >>>>>>>>>>>>>>>>>>>> performance
> >>>>>>>>>>>>>>>>>>>>>>>>>>> impact. Atomic Checkpointing is definitely
> >> the
> >>>>>> "best"
> >>>>>>>>>>>>>>>> solution,
> >>>>>>>>>>>>>>>>>>>> but
> >>>>>>>>>>>>>>>>>>>>>>>> I'm not
> >>>>>>>>>>>>>>>>>>>>>>>>>>> sure if this is enough to bring it back
> >> into
> >>>> this
> >>>>>> KIP.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> I'm currently working on moving all the
> >>>>>> transactional
> >>>>>>>>>>>> logic
> >>>>>>>>>>>>>>>>>>>> directly
> >>>>>>>>>>>>>>>>>>>>>>>> into
> >>>>>>>>>>>>>>>>>>>>>>>>>>> RocksDBStore itself, which does away with
> >> the
> >>>>>>>>>>>>>>>>>>>>>> StateStore#newTransaction
> >>>>>>>>>>>>>>>>>>>>>>>>>>> method, and reduces the number of new
> >> classes
> >>>>>>>> introduced,
> >>>>>>>>>>>>>>>>>>>>>> significantly
> >>>>>>>>>>>>>>>>>>>>>>>>>>> reducing the complexity. If it works, and
> >> the
> >>>>>>>> complexity
> >>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>> drastically
> >>>>>>>>>>>>>>>>>>>>>>>>>>> reduced, I may try bringing back Atomic
> >>>> Checkpoints
> >>>>>>>> into
> >>>>>>>>>>>>>>>> this
> >>>>>>>>>>>>>>>>>> KIP.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>> On Mon, 3 Jul 2023 at 15:27, Bruno Cadonna
> >> <
> >>>>>>>>>>>>>>>> cado...@apache.org>
> >>>>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Nick,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Thanks for the insights! Very interesting!
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> As far as I understand, you want to
> >> atomically
> >>>>>> update
> >>>>>>>>>> the
> >>>>>>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> from the transaction buffer, flush the
> >>>> memtable
> >>>>>> of a
> >>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> write the checkpoint not after the commit
> >> time
> >>>>>> elapsed
> >>>>>>>>>>>> but
> >>>>>>>>>>>>>>>>>> after
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> transaction buffer reached a size that
> >> would
> >>>> lead
> >>>>>> to
> >>>>>>>>>>>>>>>> exceeding
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> statestore.transaction.buffer.max.bytes
> >>>> before the
> >>>>>>>> next
> >>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>>>>> interval
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> ends.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> That means, the Kafka transaction would
> >> commit
> >>>>>> every
> >>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>> interval
> >>>>>>>>>>>>>>>>>>>>>>>> but
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> the state store will only be atomically
> >>>> updated
> >>>>>>>> roughly
> >>>>>>>>>>>>>>>> every
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> statestore.transaction.buffer.max.bytes of
> >>>> data.
> >>>>>> Also
> >>>>>>>> IQ
> >>>>>>>>>>>>>>>> would
> >>>>>>>>>>>>>>>>>>>> then
> >>>>>>>>>>>>>>>>>>>>>>>> only
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> see new data roughly every
> >>>>>>>>>>>>>>>>>>>> statestore.transaction.buffer.max.bytes.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> After a failure the state store needs to
> >>>> restore
> >>>>>> up to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> statestore.transaction.buffer.max.bytes.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Is this correct?
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> What about atomically updating the state
> >> store
> >>>>>> from
> >>>>>>>> the
> >>>>>>>>>>>>>>>>>>>> transaction
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> buffer every commit interval and writing
> >> the
> >>>>>>>> checkpoint
> >>>>>>>>>>>>>>>> (thus,
> >>>>>>>>>>>>>>>>>>>>>>>> flushing
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> the memtable) every configured amount of
> >> data
> >>>>>> and/or
> >>>>>>>>>>>>>>>> number of
> >>>>>>>>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> intervals? In such a way, we would have
> >> the
> >>>> same
> >>>>>> delay
> >>>>>>>>>>>> for
> >>>>>>>>>>>>>>>>>>>> records
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> appearing in output topics and IQ because
> >> both
> >>>>>> would
> >>>>>>>>>>>>>>>> appear
> >>>>>>>>>>>>>>>>>> when
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Kafka transaction is committed. However,
> >>>> after a
> >>>>>>>> failure
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> still needs to restore up to
> >>>>>>>>>>>>>>>>>>>> statestore.transaction.buffer.max.bytes
> >>>>>>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> it might restore data that is already in
> >> the
> >>>> state
> >>>>>>>> store
> >>>>>>>>>>>>>>>>>> because
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> checkpoint lags behind the last stable
> >> offset
> >>>>>> (i.e.
> >>>>>>>> the
> >>>>>>>>>>>>>>>> last
> >>>>>>>>>>>>>>>>>>>>>> committed
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> offset) of the changelog topics. Restoring
> >>>> data
> >>>>>> that
> >>>>>>>> is
> >>>>>>>>>>>>>>>> already
> >>>>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> state store is idempotent, so eos should
> >> not
> >>>>>> violated.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> This solution needs at least one new
> >> config to
> >>>>>> specify
> >>>>>>>>>>>>>>>> when a
> >>>>>>>>>>>>>>>>>>>>>>>> checkpoint
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> should be written.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> A small correction to your previous e-mail
> >>>> that
> >>>>>> does
> >>>>>>>> not
> >>>>>>>>>>>>>>>> change
> >>>>>>>>>>>>>>>>>>>>>>>> anything
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> you said: Under alos the default commit
> >>>> interval
> >>>>>> is 30
> >>>>>>>>>>>>>>>> seconds,
> >>>>>>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>>>>> five
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> seconds.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Bruno
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> On 01.07.23 12:37, Nick Telford wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi everyone,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> I've begun performance testing my branch
> >> on
> >>>> our
> >>>>>>>> staging
> >>>>>>>>>>>>>>>>>>>>>> environment,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> putting it through its paces in our
> >>>> non-trivial
> >>>>>>>>>>>>>>>> application.
> >>>>>>>>>>>>>>>>>> I'm
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> already
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> observing the same increased flush rate
> >> that
> >>>> we
> >>>>>> saw
> >>>>>>>> the
> >>>>>>>>>>>>>>>> last
> >>>>>>>>>>>>>>>>>>>> time
> >>>>>>>>>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> attempted to use a version of this KIP,
> >> but
> >>>> this
> >>>>>>>> time,
> >>>>>>>>>> I
> >>>>>>>>>>>>>>>>>> think I
> >>>>>>>>>>>>>>>>>>>>>> know
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> why.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Pre-KIP-892, StreamTask#postCommit,
> >> which is
> >>>>>> called
> >>>>>>>> at
> >>>>>>>>>>>>>>>> the end
> >>>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Task
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> commit process, has the following
> >> behaviour:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> - Under ALOS: checkpoint the
> >> state
> >>>>>> stores.
> >>>>>>>>>> This
> >>>>>>>>>>>>>>>>>> includes
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> flushing memtables in RocksDB.
> >>>> This is
> >>>>>>>>>>>> acceptable
> >>>>>>>>>>>>>>>>>>>> because the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> default
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> commit.interval.ms is 5
> >> seconds,
> >>>> so
> >>>>>>>> forcibly
> >>>>>>>>>>>>>>>> flushing
> >>>>>>>>>>>>>>>>>>>>>> memtables
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> every 5
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> seconds is acceptable for most
> >>>>>>>> applications.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> - Under EOS: checkpointing is
> >> not
> >>>> done,
> >>>>>>>>>> *unless*
> >>>>>>>>>>>>>>>> it's
> >>>>>>>>>>>>>>>>>>>> being
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> forced, due
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> to e.g. the Task closing or
> >> being
> >>>>>> revoked.
> >>>>>>>>>> This
> >>>>>>>>>>>>>>>> means
> >>>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>> under
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> normal
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> processing conditions, the
> >> state
> >>>> stores
> >>>>>>>> will
> >>>>>>>>>> not
> >>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>>> checkpointed,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> and will
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> not have memtables flushed at
> >> all ,
> >>>>>> unless
> >>>>>>>>>>>> RocksDB
> >>>>>>>>>>>>>>>>>>>> decides to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> flush them on
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> its own. Checkpointing stores
> >> and
> >>>>>>>>>> force-flushing
> >>>>>>>>>>>>>>>> their
> >>>>>>>>>>>>>>>>>>>>>> memtables
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> is only
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> done when a Task is being
> >> closed.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Under EOS, KIP-892 needs to checkpoint
> >>>> stores on
> >>>>>> at
> >>>>>>>>>>>> least
> >>>>>>>>>>>>>>>>>> *some*
> >>>>>>>>>>>>>>>>>>>>>>>> normal
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Task commits, in order to write the
> >> RocksDB
> >>>>>>>> transaction
> >>>>>>>>>>>>>>>>>> buffers
> >>>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> state stores, and to ensure the offsets
> >> are
> >>>>>> synced to
> >>>>>>>>>>>>>>>> disk to
> >>>>>>>>>>>>>>>>>>>>>> prevent
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> restores from getting out of hand.
> >>>> Consequently,
> >>>>>> my
> >>>>>>>>>>>>>>>> current
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> implementation
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> calls maybeCheckpoint on *every* Task
> >> commit,
> >>>>>> which
> >>>>>>>> is
> >>>>>>>>>>>>>>>> far too
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> frequent.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> This causes checkpoints every 10,000
> >> records,
> >>>>>> which
> >>>>>>>> is
> >>>>>>>>>> a
> >>>>>>>>>>>>>>>>>> change
> >>>>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> flush
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> behaviour, potentially causing
> >> performance
> >>>>>> problems
> >>>>>>>> for
> >>>>>>>>>>>>>>>> some
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> applications.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> I'm looking into possible solutions, and
> >> I'm
> >>>>>>>> currently
> >>>>>>>>>>>>>>>> leaning
> >>>>>>>>>>>>>>>>>>>>>>>> towards
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> using the
> >>>> statestore.transaction.buffer.max.bytes
> >>>>>>>>>>>>>>>>>> configuration
> >>>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> checkpoint Tasks once we are likely to
> >>>> exceed it.
> >>>>>>>> This
> >>>>>>>>>>>>>>>> would
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> complement the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> existing "early Task commit"
> >> functionality
> >>>> that
> >>>>>> this
> >>>>>>>>>>>>>>>>>>>> configuration
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> provides, in the following way:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> - Currently, we use
> >>>>>>>>>>>>>>>>>>>> statestore.transaction.buffer.max.bytes
> >>>>>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> force an
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> early Task commit if processing
> >>>> more
> >>>>>>>> records
> >>>>>>>>>>>> would
> >>>>>>>>>>>>>>>>>> cause
> >>>>>>>>>>>>>>>>>>>> our
> >>>>>>>>>>>>>>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> transactions to exceed the
> >> memory
> >>>>>> assigned
> >>>>>>>> to
> >>>>>>>>>>>>>>>> them.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> - New functionality: when a
> >> Task
> >>>> *does*
> >>>>>>>>>> commit,
> >>>>>>>>>>>>>>>> we will
> >>>>>>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> checkpoint
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> the stores (and hence flush the
> >>>>>> transaction
> >>>>>>>>>>>>>>>> buffers)
> >>>>>>>>>>>>>>>>>>>> unless
> >>>>>>>>>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> expect to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> cross the
> >>>>>>>>>>>> statestore.transaction.buffer.max.bytes
> >>>>>>>>>>>>>>>>>>>> threshold
> >>>>>>>>>>>>>>>>>>>>>>>> before
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> the next
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> commit
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> I'm also open to suggestions.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> On Thu, 22 Jun 2023 at 14:06, Nick
> >> Telford <
> >>>>>>>>>>>>>>>>>>>> nick.telf...@gmail.com
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Bruno!
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 3.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> By "less predictable for users", I
> >> meant in
> >>>>>> terms of
> >>>>>>>>>>>>>>>>>>>> understanding
> >>>>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> performance profile under various
> >>>>>> circumstances. The
> >>>>>>>>>>>>>>>> more
> >>>>>>>>>>>>>>>>>>>> complex
> >>>>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> solution, the more difficult it would
> >> be for
> >>>>>> users
> >>>>>>>> to
> >>>>>>>>>>>>>>>>>>>> understand
> >>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> performance they see. For example,
> >> spilling
> >>>>>> records
> >>>>>>>> to
> >>>>>>>>>>>>>>>> disk
> >>>>>>>>>>>>>>>>>>>> when
> >>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> transaction buffer reaches a threshold
> >>>> would, I
> >>>>>>>>>> expect,
> >>>>>>>>>>>>>>>>>> reduce
> >>>>>>>>>>>>>>>>>>>>>> write
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> throughput. This reduction in write
> >>>> throughput
> >>>>>> could
> >>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>> unexpected,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> potentially difficult to
> >>>> diagnose/understand for
> >>>>>>>>>> users.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> At the moment, I think the "early
> >> commit"
> >>>>>> concept is
> >>>>>>>>>>>>>>>>>> relatively
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> straightforward; it's easy to document,
> >> and
> >>>>>>>>>>>> conceptually
> >>>>>>>>>>>>>>>>>> fairly
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> obvious to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> users. We could probably add a metric to
> >>>> make it
> >>>>>>>>>> easier
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>> understand
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> when
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> it happens though.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 3. (the second one)
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> The IsolationLevel is *essentially* an
> >>>> indirect
> >>>>>> way
> >>>>>>>> of
> >>>>>>>>>>>>>>>>>> telling
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> StateStores
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> whether they should be transactional.
> >>>>>> READ_COMMITTED
> >>>>>>>>>>>>>>>>>>>> essentially
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> requires
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> transactions, because it dictates that
> >> two
> >>>>>> threads
> >>>>>>>>>>>>>>>> calling
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> `newTransaction()` should not see writes
> >>>> from
> >>>>>> the
> >>>>>>>>>> other
> >>>>>>>>>>>>>>>>>>>>>> transaction
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> until
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> they have been committed. With
> >>>>>> READ_UNCOMMITTED, all
> >>>>>>>>>>>>>>>> bets are
> >>>>>>>>>>>>>>>>>>>> off,
> >>>>>>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> stores can allow threads to observe
> >> written
> >>>>>> records
> >>>>>>>> at
> >>>>>>>>>>>>>>>> any
> >>>>>>>>>>>>>>>>>>>> time,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> which is
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> essentially "no transactions". That
> >> said,
> >>>>>>>> StateStores
> >>>>>>>>>>>>>>>> are
> >>>>>>>>>>>>>>>>>> free
> >>>>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> implement
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> these guarantees however they can,
> >> which is
> >>>> a
> >>>>>> bit
> >>>>>>>> more
> >>>>>>>>>>>>>>>>>> relaxed
> >>>>>>>>>>>>>>>>>>>>>> than
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> dictating "you must use transactions".
> >> For
> >>>>>> example,
> >>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>>>> RocksDB
> >>>>>>>>>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> would
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> implement these as READ_COMMITTED ==
> >>>> WBWI-based
> >>>>>>>>>>>>>>>>>> "transactions",
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> READ_UNCOMMITTED == direct writes to the
> >>>>>> database.
> >>>>>>>> But
> >>>>>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>>>> other
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> storage
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> engines, it might be preferable to
> >> *always*
> >>>> use
> >>>>>>>>>>>>>>>> transactions,
> >>>>>>>>>>>>>>>>>>>> even
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> when
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> unnecessary; or there may be storage
> >> engines
> >>>>>> that
> >>>>>>>>>> don't
> >>>>>>>>>>>>>>>>>> provide
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> transactions, but the isolation
> >> guarantees
> >>>> can
> >>>>>> be
> >>>>>>>> met
> >>>>>>>>>>>>>>>> using a
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> different
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> technique.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> My idea was to try to keep the
> >> StateStore
> >>>>>> interface
> >>>>>>>> as
> >>>>>>>>>>>>>>>>>> loosely
> >>>>>>>>>>>>>>>>>>>>>>>> coupled
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> from the Streams engine as possible, to
> >> give
> >>>>>>>>>>>>>>>> implementers
> >>>>>>>>>>>>>>>>>> more
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> freedom, and
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> reduce the amount of internal knowledge
> >>>>>> required.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> That said, I understand that
> >>>> "IsolationLevel"
> >>>>>> might
> >>>>>>>>>> not
> >>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> right
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> abstraction, and we can always make it
> >> much
> >>>> more
> >>>>>>>>>>>>>>>> explicit if
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> required, e.g.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> boolean transactional()
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 7-8.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I can make these changes either later
> >> today
> >>>> or
> >>>>>>>>>>>> tomorrow.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Small update:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I've rebased my branch on trunk and
> >> fixed a
> >>>>>> bunch of
> >>>>>>>>>>>>>>>> issues
> >>>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> needed
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> addressing. Currently, all the tests
> >> pass,
> >>>>>> which is
> >>>>>>>>>>>>>>>>>> promising,
> >>>>>>>>>>>>>>>>>>>> but
> >>>>>>>>>>>>>>>>>>>>>>>> it
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> will
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> need to undergo some performance
> >> testing. I
> >>>>>> haven't
> >>>>>>>>>>>>>>>> (yet)
> >>>>>>>>>>>>>>>>>>>> worked
> >>>>>>>>>>>>>>>>>>>>>> on
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> removing the `newTransaction()` stuff,
> >> but I
> >>>>>> would
> >>>>>>>>>>>>>>>> expect
> >>>>>>>>>>>>>>>>>> that,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> behaviourally, it should make no
> >>>> difference. The
> >>>>>>>>>> branch
> >>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>> available
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> at
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>> https://github.com/nicktelford/kafka/tree/KIP-892-c
> >>>>>>>>>> if
> >>>>>>>>>>>>>>>>>> anyone
> >>>>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> interested in taking an early look.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> On Thu, 22 Jun 2023 at 11:59, Bruno
> >> Cadonna
> >>>> <
> >>>>>>>>>>>>>>>>>>>> cado...@apache.org>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Nick,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 1.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Yeah, I agree with you. That was
> >> actually
> >>>> also
> >>>>>> my
> >>>>>>>>>>>>>>>> point. I
> >>>>>>>>>>>>>>>>>>>>>>>> understood
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> that John was proposing the ingestion
> >> path
> >>>> as
> >>>>>> a way
> >>>>>>>>>> to
> >>>>>>>>>>>>>>>> avoid
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> early
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> commits. Probably, I misinterpreted the
> >>>> intent.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 2.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I agree with John here, that actually
> >> it is
> >>>>>> public
> >>>>>>>>>>>>>>>> API. My
> >>>>>>>>>>>>>>>>>>>>>> question
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> how this usage pattern affects normal
> >>>>>> processing.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 3.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> My concern is that checking for the
> >> size
> >>>> of the
> >>>>>>>>>>>>>>>> transaction
> >>>>>>>>>>>>>>>>>>>>>> buffer
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> maybe triggering an early commit
> >> affects
> >>>> the
> >>>>>> whole
> >>>>>>>>>>>>>>>>>> processing
> >>>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Kafka
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Streams. The transactionality of a
> >> state
> >>>> store
> >>>>>> is
> >>>>>>>> not
> >>>>>>>>>>>>>>>>>>>> confined to
> >>>>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> state store itself, but spills over and
> >>>>>> changes the
> >>>>>>>>>>>>>>>> behavior
> >>>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>>>>> other
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> parts of the system. I agree with you
> >> that
> >>>> it
> >>>>>> is a
> >>>>>>>>>>>>>>>> decent
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> compromise. I
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> just wanted to analyse the downsides
> >> and
> >>>> list
> >>>>>> the
> >>>>>>>>>>>>>>>> options to
> >>>>>>>>>>>>>>>>>>>>>>>> overcome
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> them. I also agree with you that all
> >>>> options
> >>>>>> seem
> >>>>>>>>>>>> quite
> >>>>>>>>>>>>>>>>>> heavy
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> compared
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> with your KIP. I do not understand
> >> what you
> >>>>>> mean
> >>>>>>>> with
> >>>>>>>>>>>>>>>> "less
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> predictable
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> for users", though.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I found the discussions about the
> >>>> alternatives
> >>>>>>>> really
> >>>>>>>>>>>>>>>>>>>>>> interesting.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> But I
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> also think that your plan sounds good
> >> and
> >>>> we
> >>>>>> should
> >>>>>>>>>>>>>>>> continue
> >>>>>>>>>>>>>>>>>>>> with
> >>>>>>>>>>>>>>>>>>>>>>>> it!
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Some comments on your reply to my
> >> e-mail on
> >>>>>> June
> >>>>>>>>>> 20th:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 3.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Ah, now, I understand the reasoning
> >> behind
> >>>>>> putting
> >>>>>>>>>>>>>>>> isolation
> >>>>>>>>>>>>>>>>>>>>>> level
> >>>>>>>>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> the state store context. Thanks! Should
> >>>> that
> >>>>>> also
> >>>>>>>> be
> >>>>>>>>>> a
> >>>>>>>>>>>>>>>> way
> >>>>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>> give
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> the state store the opportunity to
> >> decide
> >>>>>> whether
> >>>>>>>> to
> >>>>>>>>>>>>>>>> turn on
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> transactions or not?
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> With my comment, I was more concerned
> >> about
> >>>>>> how do
> >>>>>>>>>> you
> >>>>>>>>>>>>>>>> know
> >>>>>>>>>>>>>>>>>>>> if a
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> checkpoint file needs to be written
> >> under
> >>>> EOS,
> >>>>>> if
> >>>>>>>> you
> >>>>>>>>>>>>>>>> do not
> >>>>>>>>>>>>>>>>>>>>>> have a
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> way
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> to know if the state store is
> >>>> transactional or
> >>>>>> not.
> >>>>>>>>>> If
> >>>>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> transactional, the checkpoint file can
> >> be
> >>>>>> written
> >>>>>>>>>>>>>>>> during
> >>>>>>>>>>>>>>>>>>>> normal
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> processing under EOS. If the state
> >> store
> >>>> is not
> >>>>>>>>>>>>>>>>>> transactional,
> >>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> checkpoint file must not be written
> >> under
> >>>> EOS.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 7.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> My point was about not only
> >> considering the
> >>>>>> bytes
> >>>>>>>> in
> >>>>>>>>>>>>>>>> memory
> >>>>>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>>>>>>>>>> config
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> statestore.uncommitted.max.bytes, but
> >> also
> >>>>>> bytes
> >>>>>>>> that
> >>>>>>>>>>>>>>>> might
> >>>>>>>>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> spilled
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> on disk. Basically, I was wondering
> >>>> whether you
> >>>>>>>>>> should
> >>>>>>>>>>>>>>>>>> remove
> >>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> "memory" in "Maximum number of memory
> >>>> bytes to
> >>>>>> be
> >>>>>>>>>> used
> >>>>>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> buffer uncommitted state-store
> >> records." My
> >>>>>>>> thinking
> >>>>>>>>>>>>>>>> was
> >>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>> even
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> if a
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> state store spills uncommitted bytes to
> >>>> disk,
> >>>>>>>>>> limiting
> >>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>> overall
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> bytes
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> might make sense. Thinking about it
> >> again
> >>>> and
> >>>>>>>>>>>>>>>> considering
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>> recent
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> discussions, it does not make too much
> >>>> sense
> >>>>>>>> anymore.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I like the name
> >>>>>>>>>>>>>>>> statestore.transaction.buffer.max.bytes that
> >>>>>>>>>>>>>>>>>>>> you
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> proposed.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 8.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> A high-level description (without
> >>>>>> implementation
> >>>>>>>>>>>>>>>> details) of
> >>>>>>>>>>>>>>>>>>>> how
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> Kafka
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Streams will manage the commit of
> >> changelog
> >>>>>>>>>>>>>>>> transactions,
> >>>>>>>>>>>>>>>>>>>> state
> >>>>>>>>>>>>>>>>>>>>>>>> store
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> transactions and checkpointing would be
> >>>> great.
> >>>>>>>> Would
> >>>>>>>>>>>> be
> >>>>>>>>>>>>>>>>>> great
> >>>>>>>>>>>>>>>>>>>> if
> >>>>>>>>>>>>>>>>>>>>>>>> you
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> could also add some sentences about the
> >>>>>> behavior in
> >>>>>>>>>>>>>>>> case of
> >>>>>>>>>>>>>>>>>> a
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> failure.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> For instance how does a transactional
> >> state
> >>>>>> store
> >>>>>>>>>>>>>>>> recover
> >>>>>>>>>>>>>>>>>>>> after a
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> failure or what happens with the
> >>>> transaction
> >>>>>>>> buffer,
> >>>>>>>>>>>>>>>> etc.
> >>>>>>>>>>>>>>>>>>>> (that
> >>>>>>>>>>>>>>>>>>>>>> is
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> what
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I meant by "fail-over" in point 9.)
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Bruno
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> On 21.06.23 18:50, Nick Telford wrote:
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Bruno,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 1.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Isn't this exactly the same issue that
> >>>>>>>>>>>>>>>> WriteBatchWithIndex
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> transactions
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> have, whereby exceeding (or likely to
> >>>> exceed)
> >>>>>>>>>>>>>>>> configured
> >>>>>>>>>>>>>>>>>>>> memory
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> needs to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> trigger an early commit?
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 2.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> This is one of my big concerns.
> >>>> Ultimately,
> >>>>>> any
> >>>>>>>>>>>>>>>> approach
> >>>>>>>>>>>>>>>>>>>> based
> >>>>>>>>>>>>>>>>>>>>>> on
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> cracking
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> open RocksDB internals and using it in
> >>>> ways
> >>>>>> it's
> >>>>>>>> not
> >>>>>>>>>>>>>>>> really
> >>>>>>>>>>>>>>>>>>>>>>>> designed
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> for is
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> likely to have some unforseen
> >> performance
> >>>> or
> >>>>>>>>>>>>>>>> consistency
> >>>>>>>>>>>>>>>>>>>> issues.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 3.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> What's your motivation for removing
> >> these
> >>>>>> early
> >>>>>>>>>>>>>>>> commits?
> >>>>>>>>>>>>>>>>>>>> While
> >>>>>>>>>>>>>>>>>>>>>> not
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> ideal, I
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> think they're a decent compromise to
> >>>> ensure
> >>>>>>>>>>>>>>>> consistency
> >>>>>>>>>>>>>>>>>>>> whilst
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> maintaining
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> good and predictable performance.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> All 3 of your suggested ideas seem
> >> *very*
> >>>>>>>>>>>>>>>> complicated, and
> >>>>>>>>>>>>>>>>>>>> might
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> actually
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> make behaviour less predictable for
> >> users
> >>>> as a
> >>>>>>>>>>>>>>>> consequence.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I'm a bit concerned that the scope of
> >> this
> >>>>>> KIP is
> >>>>>>>>>>>>>>>> growing a
> >>>>>>>>>>>>>>>>>>>> bit
> >>>>>>>>>>>>>>>>>>>>>>>> out
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> of
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> control. While it's good to discuss
> >> ideas
> >>>> for
> >>>>>>>> future
> >>>>>>>>>>>>>>>>>>>>>>>> improvements, I
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> think
> >>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> it's important to narrow the scope
> >> down
> >>>> to a
> >>>>>>>> design
> >>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>> achieves
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> most
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> pressing objectives (constant sized
> >>>>>> restorations
> >>>>>>>>>>>>>>>> during
> >>>>>>>>>>>>>>>>>> dirty
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> close/unexpected errors). Any design
> >> that
> >>>>>> this KIP
> >>>>>>>>>>>>>>>> produces
> >>>>>>>>>>>>>>>>>>>> can
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> ultimately
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> be changed in the future, especially
> >> if
> >>>> the
> >>>>>> bulk
> >>>>>>>> of
> >>>>>>>>>>>>>>>> it is
> >>>>>>>>>>>>>>>>>>>>>> internal
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> behaviour.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I'm going to spend some time next week
> >>>> trying
> >>>>>> to
> >>>>>>>>>>>>>>>> re-work
> >>>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> original
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> WriteBatchWithIndex design to remove
> >> the
> >>>>>>>>>>>>>>>> newTransaction()
> >>>>>>>>>>>>>>>>>>>>>> method,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> such
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> that
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> it's just an implementation detail of
> >>>>>>>> RocksDBStore.
> >>>>>>>>>>>>>>>> That
> >>>>>>>>>>>>>>>>>>>> way, if
> >>>>>>>>>>>>>>>>>>>>>>>> we
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> want to
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> replace WBWI with something in the
> >> future,
> >>>>>> like
> >>>>>>>> the
> >>>>>>>>>>>>>>>> SST
> >>>>>>>>>>>>>>>>>> file
> >>>>>>>>>>>>>>>>>>>>>>>>>>>> management
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> outlined by John, then we can do so
> >> with
> >>>>>> little/no
> >>>>>>>>>>>> API
> >>>>>>>>>>>>>>>>>>>> changes.
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Nick
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>
> >>>>>>
> >>>>
> >>
> >
