Hi Matthias, I like it. I've updated the KIP to reflect that detail; I put the details in the docs for pause.
Cheers, Jim On Tue, May 10, 2022 at 7:51 PM Matthias J. Sax <mj...@apache.org> wrote: > Thanks for the KIP. Overall LGTM. > > Can we clarify one question: would it be allowed to call `pause()` > before calling `start()`? I don't see any reason why we would need to > disallow it? > > It could be helpful to start a KafkaStreams client in paused state -- > otherwise there is a race between calling `start()` and calling `pause()`. > > If we allow it, we should clearly document it. > > > -Matthias > > On 5/10/22 12:04 PM, Jim Hughes wrote: > > Hi Bill, all, > > > > Thank you. I've updated the KIP to reflect pausing standby tasks as > well. > > I think all the outstanding points have been addressed and I'm going to > > start the vote thread! > > > > Cheers, > > > > Jim > > > > > > > > On Tue, May 10, 2022 at 2:43 PM Bill Bejeck <bbej...@gmail.com> wrote: > > > >> Hi Jim, > >> > >> After reading the comments on the KIP, I agree that it makes sense to > pause > >> all activities and any changes can be made later on. > >> > >> Thanks, > >> Bill > >> > >> On Tue, May 10, 2022 at 4:03 AM Bruno Cadonna <cado...@apache.org> > wrote: > >> > >>> Hi Jim, > >>> > >>> Thanks for the KIP! > >>> > >>> I am fine with the KIP in general. > >>> > >>> However, I am with Sophie and John to also pause the standbys for the > >>> reasons they brought up. Is there a specific reason you want to keep > >>> standbys going? It feels like premature optimization to me. We can > still > >>> add keeping standby running in a follow up if needed. > >>> > >>> Best, > >>> Bruno > >>> > >>> On 10.05.22 05:15, Sophie Blee-Goldman wrote: > >>>> Thanks Jim, just one note/question on the standby tasks: > >>>> > >>>> At the minute, my moderately held position is that standby tasks ought > >> to > >>>>> continue reading and remain caught up. If standby tasks would run > out > >>> of > >>>>> space, there are probably bigger problems. > >>>> > >>>> > >>>> For a single node application, or when the #pause API is invoked on > all > >>>> instances, > >>>> then there won't be any further active processing and thus nothing to > >>> keep > >>>> up with, > >>>> right? So for that case, it's just a matter of whether any standbys > >> that > >>>> are lagging > >>>> will have the chance to catch up to the (paused) active task state > >> before > >>>> they stop > >>>> as well, in which case having them continue feels fine to me. However > >>> this > >>>> is a > >>>> relatively trivial benefit and I would only consider it as a deciding > >>>> factor when all > >>>> things are equal otherwise. > >>>> > >>>> My concern is the more interesting case: when this feature is used to > >>> pause > >>>> only > >>>> one nodes, or some subset of the overall application. In this case, > >> yes, > >>>> the standby > >>>> tasks will indeed fall out of sync. But the only reason I can imagine > >>>> someone using > >>>> the pause feature in such a way is because there is something going > >>> wrong, > >>>> or about > >>>> to go wrong, on that particular node. For example as mentioned above, > >> if > >>>> the user > >>>> wants to cut down on costs without stopping everything, or if the node > >> is > >>>> about to > >>>> run out of disk or needs to be debugged or so on. And in this case, > >>>> continuing to > >>>> process the standby tasks while other instances continue to run would > >>>> pretty much > >>>> defeat the purpose of pausing it entirely, and might have unpleasant > >>>> consequences > >>>> for the unsuspecting developer. > >>>> > >>>> All that said, I don't want to block this KIP so if you have strong > >>>> feelings about the > >>>> standby behavior I'm happy to back down. I'm only pushing back now > >>> because > >>>> it > >>>> felt like there wasn't any particular motivation for the standbys to > >>>> continue processing > >>>> or not, and I figured I'd try to fill in this gap with my thoughts on > >> the > >>>> matter :) > >>>> Either way we should just make sure that this behavior is documented > >>>> clearly, > >>>> since it may be surprising if we decide to only pause active > processing > >>>> (another option > >>>> is to rename the method something like #pauseProcessing or > >>>> #pauseActiveProcessing > >>>> so that it's hard to miss). > >>>> > >>>> Thanks! Sorry for the lengthy response, but hopefully we won't need to > >>>> debate this any > >>>> further. Beyond this I'm satisfied with the latest proposal > >>>> > >>>> On Mon, May 9, 2022 at 5:16 PM John Roesler <vvcep...@apache.org> > >> wrote: > >>>> > >>>>> Thanks for the updates, Jim! > >>>>> > >>>>> After this discussion and your updates, this KIP looks good to me. > >>>>> > >>>>> Thanks, > >>>>> John > >>>>> > >>>>> On Mon, May 9, 2022, at 17:52, Jim Hughes wrote: > >>>>>> Hi Sophie, all, > >>>>>> > >>>>>> I've updated the KIP with feedback from the discussion so far: > >>>>>> > >>>>> > >>> > >> > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=211882832 > >>>>>> > >>>>>> As a terse summary of my current position: > >>>>>> Pausing will only stop processing and punctuation (respecting > modular > >>>>>> topologies). > >>>>>> Paused topologies will still a) consume from input topics, b) call > >> the > >>>>>> usual commit pathways (commits will happen basically as they would > >>> have), > >>>>>> and c) standBy tasks will still be processed. > >>>>>> > >>>>>> Shout if the KIP or those details still need some TLC. Responding > to > >>>>>> Sophie inline below. > >>>>>> > >>>>>> > >>>>>> On Mon, May 9, 2022 at 6:06 PM Sophie Blee-Goldman > >>>>>> <sop...@confluent.io.invalid> wrote: > >>>>>> > >>>>>>> Don't worry, I'm going to be adding the APIs for topology-level > >>> pausing > >>>>> as > >>>>>>> part of the modular topologies KIP, > >>>>>>> so we don't need to worry about that for now. That said, I don't > >> think > >>>>> we > >>>>>>> should brush it off entirely and design > >>>>>>> this feature in a way that's going to be incompatible or hugely > >> raise > >>>>> the > >>>>>>> LOE on bringing the (mostly already > >>>>>>> implemented) modular topologies feature into the public API, just > >>>>>>> because it "won the race to write a KIP" :) > >>>>>>> > >>>>>> > >>>>>> Yes, I'm hoping that this is all compatible with modular > >> topologies. I > >>>>>> haven't seen anything so far which seems to be a problem; this KIP > is > >>>>> just > >>>>>> in a weird state to discuss details of acting on modular > >> topologies.:) > >>>>>> > >>>>>> > >>>>>>> I may be biased (ok, I definitely am), but I'm not in favor of > >> adding > >>>>> this > >>>>>>> as a state regardless of the modular topologies. > >>>>>>> First of all any change to the KafkaStreams state machine is a > >>> breaking > >>>>>>> change, no? So we would have to wait until > >>>>>>> the next major release which seems like an unnecessary thing to > >> block > >>>>> on. > >>>>>>> (Whether to add this as a state to the > >>>>>>> StreamThread's FSM is an implementation detail). > >>>>>>> > >>>>>> > >>>>>> +1. I am sold on skipping out on new states. I had that as a > >> rejected > >>>>>> alternative in the KIP and have added a few more words to that bit. > >>>>>> > >>>>>> > >>>>>>> Also, the semantics of using an `isPaused` method to distinguish a > >>>>> paused > >>>>>>> instance (or topology) make more sense > >>>>>>> to me -- this is a user-specified status, whereas the KafkaStreams > >>>>> state is > >>>>>>> intended to relay the status of the system > >>>>>>> itself. For example, if we are going to continue to poll during > >> pause, > >>>>> then > >>>>>>> shouldn't the client transition to REBALANCING? > >>>>>>> I believe it makes sense to still allow distinguishing these states > >>>>> while a > >>>>>>> client is paused, whereas making PAUSED its > >>>>>>> own state means you can't tell when the client is rebalancing vs > >>>>> running, > >>>>>>> or whether it is paused or dead: presumably > >>>>>>> the NOT_RUNNING/ERROR state would trump the PAUSED state, which > >> means > >>>>> you > >>>>>>> would not be able to rely on > >>>>>>> checking the state to see if you had called PAUSED on that > instance. > >>>>>>> Obviously you can work around this by just > >>>>>>> maintaining a flag in the usercode, but all this feels very > >> unnatural > >>>>> to me > >>>>>>> vs just checking the `#isPaused` API. > >>>>>>> > >>>>>>> On that note, I had one question -- at what point would the > >>> `#isPaused` > >>>>>>> check return true? Would it do so immediately > >>>>>>> after pausing the instance, or only once it has finished committing > >>>>> offsets > >>>>>>> and stopped returning records? > >>>>>>> > >>>>>> > >>>>>> Immediately, `#isPaused` tells you about metadata. > >>>>>> > >>>>>> > >>>>>>> Finally, on the note of punctuators I think it would make most > sense > >>> to > >>>>>>> either pause these as well or else add this an > >>>>>>> an explicit option for the user. If this feature is used to, for > >>>>> example, > >>>>>>> help save on processing costs while an app is > >>>>>>> not in use, then it would probably be surprising and perhaps > >> alarming > >>> to > >>>>>>> see certain kinds of processing still continue. > >>>>>>> > >>>>>> > >>>>>> From other parts of the discussion, I'm sold on pausing > punctuation. > >>>>>> > >>>>>> > >>>>>>> The question of whether to continue fetching for standby tasks is > >>> maybe > >>>>> a > >>>>>>> bit more debatable, as it would certainly be > >>>>>>> nice to find your clients all caught up when you go to resume the > >>>>> instance > >>>>>>> again, but I would still strongly suggest > >>>>>>> pausing these as well. To use a similar example, imagine if you > >> paused > >>>>> an > >>>>>>> app because it was about to run out of > >>>>>>> disk. If the standbys kept processing and filled up the remaining > >>> space, > >>>>>>> you'd probably feel a bit betrayed by this API. > >>>>>>> > >>>>>>> WDYT? > >>>>>>> > >>>>>> > >>>>>> At the minute, my moderately held position is that standby tasks > >> ought > >>> to > >>>>>> continue reading and remain caught up. If standby tasks would run > >> out > >>> of > >>>>>> space, there are probably bigger problems. > >>>>>> > >>>>>> If later it is desirable to manage punctuation or standby tasks, > then > >>> it > >>>>>> should be easy for future folks to modify things. > >>>>>> > >>>>>> Overall, I'd frame this KIP as "pause processing resulting in > >> outputs". > >>>>>> > >>>>>> Cheers, > >>>>>> > >>>>>> Jim > >>>>>> > >>>>>> > >>>>>> > >>>>>>> On Mon, May 9, 2022 at 10:33 AM Guozhang Wang <wangg...@gmail.com> > >>>>> wrote: > >>>>>>> > >>>>>>>> I think for named topology we can leave the scope of this KIP as > >> "all > >>>>> or > >>>>>>>> nothing", i.e. when you pause an instance you pause all of its > >>>>>>> topologies. > >>>>>>>> I raised this question in my previous email just trying to clarify > >> if > >>>>>>> this > >>>>>>>> is what you have in mind. We can leave the question of finer > >>>>> controlled > >>>>>>>> pausing behavior for later when we have named topology being > >> exposed > >>>>> via > >>>>>>>> another KIP. > >>>>>>>> > >>>>>>>> > >>>>>>>> Guozhang > >>>>>>>> > >>>>>>>> On Mon, May 9, 2022 at 7:50 AM John Roesler <vvcep...@apache.org> > >>>>> wrote: > >>>>>>>> > >>>>>>>>> Hi Jim, > >>>>>>>>> > >>>>>>>>> Thanks for the replies. This all sounds good to me. Just two > >> further > >>>>>>>>> comments: > >>>>>>>>> > >>>>>>>>> 3. It seems like you should aim for the simplest semantics. If > the > >>>>>>> intent > >>>>>>>>> is to “pause” the instance, then you’d better pause the whole > >>>>> instance. > >>>>>>>> If > >>>>>>>>> you leave punctuations and standbys running, I expect we’d see > bug > >>>>>>>> reports > >>>>>>>>> come in that the instance isn’t really paused. > >>>>>>>>> > >>>>>>>>> 5. Since you won the race to write a KIP, I don’t think it makes > >> too > >>>>>>> much > >>>>>>>>> sense to worry too much about modular topologies. When they > >> propose > >>>>>>> their > >>>>>>>>> KIP, they will have to specify a lot of state management > behavior, > >>>>> and > >>>>>>>>> pause/resume will have to be part of it. If they have some > concern > >>>>>>> about > >>>>>>>>> your KIP, they’ll chime in. It doesn’t make sense for you to try > >> and > >>>>>>>> guess > >>>>>>>>> what that proposal will look like. > >>>>>>>>> > >>>>>>>>> To be honest, you’re proposing a KafkaStreams runtime-level > >>>>>>> pause/resume > >>>>>>>>> function, not a topology-level one anyway, so it seems pretty > >> clear > >>>>>>> that > >>>>>>>> it > >>>>>>>>> would pause the whole runtime (of a single instance) regardless > of > >>>>> any > >>>>>>>>> modular topologies. If the intent is to pause individual > >> topologies > >>>>> in > >>>>>>>> the > >>>>>>>>> future, you’d need a different API anyway. > >>>>>>>>> > >>>>>>>>> Thanks! > >>>>>>>>> -John > >>>>>>>>> > >>>>>>>>> On Mon, May 9, 2022, at 08:10, Jim Hughes wrote: > >>>>>>>>>> Hi John, > >>>>>>>>>> > >>>>>>>>>> Long emails are great; responding inline! > >>>>>>>>>> > >>>>>>>>>> On Sat, May 7, 2022 at 4:54 PM John Roesler < > vvcep...@apache.org > >>> > >>>>>>>> wrote: > >>>>>>>>>> > >>>>>>>>>>> Thanks for the KIP, Jim! > >>>>>>>>>>> > >>>>>>>>>>> This conversation seems to highlight that the KIP needs to > >>>>> specify > >>>>>>>>>>> some of its behavior as well as its APIs, where the behavior is > >>>>>>>>>>> observable and significant to users. > >>>>>>>>>>> > >>>>>>>>>>> For example: > >>>>>>>>>>> > >>>>>>>>>>> 1. Do you plan to have a guarantee that immediately after > >>>>>>>>>>> calling KafkaStreams.pause(), users should observe that the > >>>>> instance > >>>>>>>>>>> stops processing new records? Or should they expect that the > >>>>> threads > >>>>>>>>>>> will continue to process some records and pause asynchronously > >>>>>>>>>>> (you already answered this in the thread earlier)? > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> I'm happy to build up to a guarantee of sorts. My current idea > >> is > >>>>>>> that > >>>>>>>>>> pause() does not do anything "exceptional" to get control back > >>>>> from a > >>>>>>>>>> running topology. A currently running topology would get to > >>>>> complete > >>>>>>>> its > >>>>>>>>>> loop. > >>>>>>>>>> > >>>>>>>>>> Separately, I'm still piecing together how commits work. By > some > >>>>>>>>>> mechanism, after a pause, I do agree that the topology needs to > >>>>>>> commit > >>>>>>>>> its > >>>>>>>>>> work in some manner. > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>>> 2. Will the threads continue to poll new records until they > >>>>>>> naturally > >>>>>>>>> fill > >>>>>>>>>>> up the task buffers, or will they immediately pause their > >>>>> Consumers > >>>>>>>>>>> as well? > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> Presently, I'm suggesting that consumers would fill up their > >>>>> buffers. > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>>> 3. Will threads continue to call (system time) punctuators, or > >>>>> would > >>>>>>>>>>> punctuations also be paused? > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> In my first pass at thinking through this, I left the > punctuators > >>>>>>>>> running. > >>>>>>>>>> To be honest, I'm not sure what they do, so my approach is > either > >>>>>>> lucky > >>>>>>>>> and > >>>>>>>>>> correct or it could be Very Clearly Wrong.;) > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>>> I realize that some of those questions simply may not have > >>>>> occurred > >>>>>>> to > >>>>>>>>>>> you, so this is not a criticism for leaving them off; I'm just > >>>>>>>> pointing > >>>>>>>>> out > >>>>>>>>>>> that although we don't tend to mention implementation details > in > >>>>>>> KIPs, > >>>>>>>>>>> we also can't be too high level, since there are a lot of > >>>>>>> operational > >>>>>>>>>>> details that users rely on to achieve various behaviors in > >>>>> Streams. > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> Ayup, I will add some details as we iron out the guarantees, > >>>>>>>>> implementation > >>>>>>>>>> details that are at the API level. This one is tough since > >>>>> internal > >>>>>>>>>> features like NamedTopologies are part of the discussion. > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>>> A couple more comments: > >>>>>>>>>>> > >>>>>>>>>>> 4. +1 to what Guozhang said. It seems like we should we also do > >> a > >>>>>>>> commit > >>>>>>>>>>> before entering the paused state. That way, any open > >> transactions > >>>>>>>> would > >>>>>>>>>>> be closed and not have to worry about timing out. Even under > >>>>> ALOS, > >>>>>>> it > >>>>>>>>>>> seems best to go ahead and complete the processing of in-flight > >>>>>>>> records > >>>>>>>>>>> by committing. That way, if anything happens to die while it's > >>>>>>> paused, > >>>>>>>>>>> existing > >>>>>>>>>>> work won't have to be repeated. Plus, if there are any > >> processors > >>>>>>> with > >>>>>>>>> side > >>>>>>>>>>> effects, users won't have to tolerate weird edge cases where a > >>>>> pause > >>>>>>>>> occurs > >>>>>>>>>>> after a processor sees a record, but before the result is sent > >> to > >>>>>>> its > >>>>>>>>>>> outputs. > >>>>>>>>>>> > >>>>>>>>>>> 5. I noticed that you proposed not to add a PAUSED state, but I > >>>>>>> didn't > >>>>>>>>>>> follow > >>>>>>>>>>> the rationale. Adding a state seems beneficial for a number of > >>>>>>>> reasons: > >>>>>>>>>>> StreamThreads already use the thread state to determine whether > >>>>> to > >>>>>>>>> process > >>>>>>>>>>> or not, so avoiding a new State would just mean adding a > >> separate > >>>>>>> flag > >>>>>>>>> to > >>>>>>>>>>> track > >>>>>>>>>>> and then checking your new flag in addition to the State in the > >>>>>>>> thread. > >>>>>>>>>>> Also, > >>>>>>>>>>> operating Streams applications is a non-trivial task, and users > >>>>> rely > >>>>>>>> on > >>>>>>>>>>> the State > >>>>>>>>>>> (and transitions) to understand Streams's behavior. Adding a > >>>>> PAUSED > >>>>>>>>> state > >>>>>>>>>>> is an elegant way to communicate to operators what is happening > >>>>> with > >>>>>>>> the > >>>>>>>>>>> application. Note that the person digging though logs and > >>>>> metrics, > >>>>>>>>> trying > >>>>>>>>>>> to understand why the application isn't doing anything is > >>>>> probably > >>>>>>> not > >>>>>>>>>>> going > >>>>>>>>>>> to be the same person who is calling pause() and resume(). > Also, > >>>>> if > >>>>>>>> you > >>>>>>>>> add > >>>>>>>>>>> a state, you don't need `isPaused()`. > >>>>>>>>>>> > >>>>>>>>>>> 5b. If you buy the arguments to go ahead and commit as well as > >>>>> the > >>>>>>>>>>> argument to add a State, then I'd also suggest to follow the > >>>>>>> existing > >>>>>>>>>>> patterns > >>>>>>>>>>> for the shutdown states by also adding PAUSING. That > >>>>>>>>>>> way, you'll also expose a way to understand that Streams > >> received > >>>>>>> the > >>>>>>>>>>> signal > >>>>>>>>>>> to pause, and that it's still processing and committing some > >>>>> records > >>>>>>>> in > >>>>>>>>>>> preparation to enter a PAUSED state. I'm not sure if a RESUMING > >>>>>>> state > >>>>>>>>> would > >>>>>>>>>>> also make sense. > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> I hit a tricky bit when thinking through having a PAUSED > >>>>> state... If > >>>>>>>> one > >>>>>>>>>> is using Named Topologies, and some of them are paused, what > >>>>> state is > >>>>>>>> the > >>>>>>>>>> Streams instance in? If we can agree on that, things may become > >>>>>>>>> clear.... > >>>>>>>>>> I can see two quick ideas: > >>>>>>>>>> > >>>>>>>>>> 1. The state is RUNNING and NamedTopologies have some other way > >>>>> to > >>>>>>>>>> indicate state. > >>>>>>>>>> > >>>>>>>>>> 2. The state is something messy like PARTIALLY_PAUSED to > reflect > >>>>>>> that > >>>>>>>>> the > >>>>>>>>>> instance has something interesting going on. > >>>>>>>>>> > >>>>>>>>>> When I poked at things initially, I did try out having different > >>>>>>>> states, > >>>>>>>>>> and I readily agree that a PAUSING state may make sense. > >>>>> (Especially > >>>>>>>> if > >>>>>>>>>> there's a need to run commits before transitioning all the way > to > >>>>>>>>> PAUSED.) > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>>> And that's all I have to say about that. I hope you don't find > >> my > >>>>>>>>>>> long message offputting. I'm fundamentally in favor of your > KIP, > >>>>>>>>>>> and I think with a little more explanation in the KIP, and a > few > >>>>>>>>>>> small tweaks to the proposal, we'll be able to provide good > >>>>>>>>>>> ergonomics to our users. > >>>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>> Thanks! > >>>>>>>>>> > >>>>>>>>>> Jim > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>>>> Thanks, > >>>>>>>>>>> -John > >>>>>>>>>>> > >>>>>>>>>>> On Sat, May 7, 2022, at 00:06, Guozhang Wang wrote: > >>>>>>>>>>>> I'm in favor of the "just pausing the instance itself“ option > >>>>> as > >>>>>>>>> well. As > >>>>>>>>>>>> for EOS, the point is that when the processing is paused, we > >>>>> would > >>>>>>>> not > >>>>>>>>>>>> trigger any `producer.send` during the time, and the > >>>>> transaction > >>>>>>>>> timeout > >>>>>>>>>>> is > >>>>>>>>>>>> sort of relying on that behavior, so my point was that it's > >>>>>>> probably > >>>>>>>>>>> better > >>>>>>>>>>>> to also commit the processing before we pause it. > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> Guozhang > >>>>>>>>>>>> > >>>>>>>>>>>> On Fri, May 6, 2022 at 6:12 PM Jim Hughes > >>>>>>>>> <jhug...@confluent.io.invalid> > >>>>>>>>>>>> wrote: > >>>>>>>>>>>> > >>>>>>>>>>>>> Hi Matthias, > >>>>>>>>>>>>> > >>>>>>>>>>>>> Since the only thing which will be paused is processing the > >>>>>>>>> topology, I > >>>>>>>>>>>>> think we can let commits happen naturally. > >>>>>>>>>>>>> > >>>>>>>>>>>>> Good point about getting the paused state to new members; it > >>>>> is > >>>>>>>>> seeming > >>>>>>>>>>>>> like the "building block" approach is a good one to keep > >>>>> things > >>>>>>>>> simple > >>>>>>>>>>> at > >>>>>>>>>>>>> first. > >>>>>>>>>>>>> > >>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>> > >>>>>>>>>>>>> Jim > >>>>>>>>>>>>> > >>>>>>>>>>>>> On Fri, May 6, 2022 at 8:31 PM Matthias J. Sax < > >>>>> mj...@apache.org > >>>>>>>> > >>>>>>>>>>> wrote: > >>>>>>>>>>>>> > >>>>>>>>>>>>>> I think it's tricky to propagate a pauseAll() via the > >>>>> rebalance > >>>>>>>>>>>>>> protocol. New members joining the group would need to get > >>>>>>> paused, > >>>>>>>>> too? > >>>>>>>>>>>>>> Could there be weird race conditions with overlapping > >>>>>>> pauseAll() > >>>>>>>>> and > >>>>>>>>>>>>>> resumeAll() calls on different instanced while there could > >>>>> be a > >>>>>>>>>>> errors / > >>>>>>>>>>>>>> network partitions or similar? > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> I would argue that similar to IQ, we provide the basic > >>>>> building > >>>>>>>>>>> blocks, > >>>>>>>>>>>>>> and leave it the user users to implement cross instance > >>>>>>>> management > >>>>>>>>>>> for a > >>>>>>>>>>>>>> pauseAll() scenario. -- Also, if there is really demand, we > >>>>> can > >>>>>>>>> always > >>>>>>>>>>>>>> add pauseAll()/resumeAll() as follow up work. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> About named typologies: I agree to Jim to not include them > >>>>> in > >>>>>>>> this > >>>>>>>>> KIP > >>>>>>>>>>>>>> as they are not a public feature yet. If we make named > >>>>>>> typologies > >>>>>>>>>>>>>> public, the corresponding KIP should extend the pause/resume > >>>>>>>>> feature > >>>>>>>>>>>>>> (ie, APIs) accordingly. Of course, the code can (and should) > >>>>>>>>> already > >>>>>>>>>>> be > >>>>>>>>>>>>>> setup to support it to be future proof. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Good call out about commit and EOS -- to simplify it, I > >>>>> think > >>>>>>> it > >>>>>>>>> might > >>>>>>>>>>>>>> be good to commit also for the at-least-once case? > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> -Matthias > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> On 5/6/22 1:05 PM, Jim Hughes wrote: > >>>>>>>>>>>>>>> Hi Bill, > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Great questions; I'll do my best to reply inline: > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> On Fri, May 6, 2022 at 3:21 PM Bill Bejeck < > >>>>>>> bbej...@gmail.com> > >>>>>>>>>>> wrote: > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Hi Jim, > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Thanks for the KIP. I have a couple of meta-questions as > >>>>>>>> well: > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> 1) Regarding pausing only a subset of running instances, > >>>>> I'm > >>>>>>>>>>> thinking > >>>>>>>>>>>>>> there > >>>>>>>>>>>>>>>> may be a use case for pausing all of them. > >>>>>>>>>>>>>>>> Would it make sense to also allow for pausing all > >>>>>>>> instances > >>>>>>>>> by > >>>>>>>>>>>>>> adding a > >>>>>>>>>>>>>>>> method `pauseAll()` or something similar? > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Honestly, I'm indifferent on this point. Presently, I > >>>>> think > >>>>>>>>> what I > >>>>>>>>>>>>> have > >>>>>>>>>>>>>>> proposed is the minimal change to get the ability to pause > >>>>>>> and > >>>>>>>>>>> resume > >>>>>>>>>>>>>>> processing. If adding a 'pauseAll()' is required, I'd be > >>>>>>> happy > >>>>>>>>> to > >>>>>>>>>>> do > >>>>>>>>>>>>>> that! > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> From Guozhang's email, it sounds like this would require > >>>>>>> using > >>>>>>>>> the > >>>>>>>>>>>>>>> rebalance protocol to trigger the coordination. Would > >>>>> there > >>>>>>> be > >>>>>>>>>>> enough > >>>>>>>>>>>>>> room > >>>>>>>>>>>>>>> in that approach to indicate that a named topology is to > >>>>> be > >>>>>>>>> paused > >>>>>>>>>>>>> across > >>>>>>>>>>>>>>> all nodes? > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> 2) Would pausing affect standby tasks? For example, > >>>>> imagine > >>>>>>>>> there > >>>>>>>>>>>>> are 3 > >>>>>>>>>>>>>>>> instances A, B, and C. > >>>>>>>>>>>>>>>> A user elects to pause instance C only but it hosts > >>>>> the > >>>>>>>>> standby > >>>>>>>>>>>>>> tasks > >>>>>>>>>>>>>>>> for A. > >>>>>>>>>>>>>>>> Would the standby tasks on the paused application > >>>>>>> continue > >>>>>>>>> to > >>>>>>>>>>> read > >>>>>>>>>>>>>> from > >>>>>>>>>>>>>>>> the changelog topic? > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Yes, standby tasks would continue reading from the > >>>>> changelog > >>>>>>>>> topic. > >>>>>>>>>>>>> All > >>>>>>>>>>>>>>> consumers would continue reading to avoid getting dropped > >>>>>>> from > >>>>>>>>> their > >>>>>>>>>>>>>>> consumer groups. > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Jim > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Thanks! > >>>>>>>>>>>>>>>> Bill > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> On Fri, May 6, 2022 at 2:44 PM Jim Hughes > >>>>>>>>>>>>> <jhug...@confluent.io.invalid > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Hi Guozhang, > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Thanks for the feedback; responses inline below: > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> On Fri, May 6, 2022 at 1:09 PM Guozhang Wang < > >>>>>>>>> wangg...@gmail.com> > >>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Hello Jim, > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Thanks for the proposed KIP. I have some meta questions > >>>>>>>> about > >>>>>>>>> it: > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> 1) Would an instance always pause/resume all of its > >>>>>>> current > >>>>>>>>> owned > >>>>>>>>>>>>>>>>>> topologies (i.e. the named topologies), or are there > >>>>> any > >>>>>>>>>>> scenarios > >>>>>>>>>>>>>>>> where > >>>>>>>>>>>>>>>>> we > >>>>>>>>>>>>>>>>>> only want to pause/resume a subset of them? > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> An instance may wish to pause some of its named > >>>>> topologies. > >>>>>>>> I > >>>>>>>>> was > >>>>>>>>>>>>>> unsure > >>>>>>>>>>>>>>>>> what to say about named topologies in the KIP since they > >>>>>>> seem > >>>>>>>>> to > >>>>>>>>>>> be > >>>>>>>>>>>>> an > >>>>>>>>>>>>>>>>> internal detail at the moment. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> I intend to add to KafkaStreamsNamedTopologyWrapper > >>>>> methods > >>>>>>>>> like: > >>>>>>>>>>>>>>>>> public void pauseNamedTopology(final String > >>>>>>>>> topologyToPause) > >>>>>>>>>>>>>>>>> public boolean isNamedTopologyPaused(final String > >>>>>>>>> topology) > >>>>>>>>>>>>>>>>> public void resumeNamedTopology(final String > >>>>>>>>>>> topologyToResume) > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> 2) From a user's perspective, do we want to always > >>>>> issue a > >>>>>>>>>>>>>>>> `pause/resume` > >>>>>>>>>>>>>>>>>> to all the instances or not? For example, we can define > >>>>>>> the > >>>>>>>>>>>>> semantics > >>>>>>>>>>>>>>>> of > >>>>>>>>>>>>>>>>>> the function as "you only need to call this function on > >>>>>>> any > >>>>>>>> of > >>>>>>>>>>> the > >>>>>>>>>>>>>>>>>> application's instances, and all instances would then > >>>>>>> pause > >>>>>>>>> (via > >>>>>>>>>>> the > >>>>>>>>>>>>>>>>>> rebalance error codes)", or as "you would call this > >>>>>>> function > >>>>>>>>> for > >>>>>>>>>>> all > >>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>> instances of an application". Which one are you > >>>>> referring > >>>>>>>> to? > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> My initial intent is that one would call this function > >>>>> on > >>>>>>> any > >>>>>>>>>>>>> instances > >>>>>>>>>>>>>>>> of > >>>>>>>>>>>>>>>>> the application that one wishes to pause. This should > >>>>>>> allow > >>>>>>>>> more > >>>>>>>>>>>>>> control > >>>>>>>>>>>>>>>>> (in case one wanted to pause a portion of the > >>>>> instances). > >>>>>>> On > >>>>>>>>> the > >>>>>>>>>>>>> other > >>>>>>>>>>>>>>>>> hand, this approach would put more work on the > >>>>> implementer > >>>>>>> to > >>>>>>>>>>>>>> coordinate > >>>>>>>>>>>>>>>>> calling pause or resume across instances. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> If the other option is more suitable, happy to do that > >>>>>>>> instead. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> 3) With EOS, there's a transaction timeout which would > >>>>>>>>> determine > >>>>>>>>>>> how > >>>>>>>>>>>>>>>>> long a > >>>>>>>>>>>>>>>>>> transaction can stay idle before it's force-aborted on > >>>>> the > >>>>>>>>> broker > >>>>>>>>>>>>>>>> side. I > >>>>>>>>>>>>>>>>>> think when a pause is issued, that means we'd need to > >>>>>>>>> immediately > >>>>>>>>>>>>>>>> commit > >>>>>>>>>>>>>>>>>> the current transaction for EOS since we do not know > >>>>> how > >>>>>>>> long > >>>>>>>>> we > >>>>>>>>>>>>> could > >>>>>>>>>>>>>>>>>> pause for. Is that right? If yes could you please > >>>>> clarify > >>>>>>>>> that in > >>>>>>>>>>>>> the > >>>>>>>>>>>>>>>> doc > >>>>>>>>>>>>>>>>>> as well. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Good point. My intent is for pause() to wait for the > >>>>> next > >>>>>>>>>>> iteration > >>>>>>>>>>>>>>>>> through `runOnce()` and then only skip over the > >>>>> processing > >>>>>>>> for > >>>>>>>>>>> paused > >>>>>>>>>>>>>>>> tasks > >>>>>>>>>>>>>>>>> in `taskManager.process(numIterations, time)`. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Do commits live inside that call or do they live > >>>>>>>>> across/outside of > >>>>>>>>>>>>> it? > >>>>>>>>>>>>>>>> In > >>>>>>>>>>>>>>>>> the former case, I think there shouldn't be any issues > >>>>> with > >>>>>>>>> EOS. > >>>>>>>>>>>>>>>>> Otherwise, we may need to work through some details to > >>>>> get > >>>>>>>> EOS > >>>>>>>>>>> right. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Once we figure that out, I can update the KIP. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Thanks, > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Jim > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Guozhang > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> On Wed, May 4, 2022 at 10:51 AM Jim Hughes > >>>>>>>>>>>>>>>> <jhug...@confluent.io.invalid > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Hi all, > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> I have written up a KIP for adding the ability to > >>>>> pause > >>>>>>> and > >>>>>>>>>>> resume > >>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>> processing of a topology in AK Streams. The KIP is > >>>>> here: > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>> > >>> > >> > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=211882832 > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Thanks in advance for your feedback! > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Jim > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> -- > >>>>>>>>>>>>>>>>>> -- Guozhang > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> > >>>>>>>>>>>> -- > >>>>>>>>>>>> -- Guozhang > >>>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> -- > >>>>>>>> -- Guozhang > >>>>>>>> > >>>>>>> > >>>>> > >>>> > >>> > >> > > >
