Hi John, Thanks for the added section, I agree with your reasoning and I think we can still use the standby replicas now.
Guozhang On Tue, Aug 20, 2019 at 3:13 PM John Roesler <j...@confluent.io> wrote: > Hi All, > > Thanks for the discussion. I've been considering the idea of giving the > "catching up" tasks a different name/role. I was in favor initially, but > after working though some details, I think it causes some problems, which > I've written up in the "rejected alternatives" part of the KIP: > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-441%3A+Smooth+Scaling+Out+for+Kafka+Streams#KIP-441:SmoothScalingOutforKafkaStreams-Addinganewkindoftask(%22moving%22,%22recovering%22,%22learner%22)fortaskmovements > > Please give it a read and let me know what you think. > > Thanks, > -John > > On Thu, Aug 8, 2019 at 7:57 PM Guozhang Wang <wangg...@gmail.com> wrote: > > > I think I agree with you Sophie. My gut feeling is that 1) it should not > be > > the major concern in assignor's algorithm for standby tasks not catching > > up, but rather be tackled in different modules, and 2) a lot of > > optimization can be down at the stream thread itself, like dedicated > > threading and larger batching, or even complicated scheduling mechanisms > > between running, restoring and standby tasks. In anyways, I think we can > > take this out of the scope of KIP-441 for now. > > > > > > Guozhang > > > > > > On Thu, Aug 8, 2019 at 5:48 PM Sophie Blee-Goldman <sop...@confluent.io> > > wrote: > > > > > > we may have other ways to not starving the standby tasks, for > example, > > by > > > > using dedicate threads for standby tasks or even consider having > > > *higher> priority for standby than active* so that we always try to > > caught > > > up standby > > > > first, then process active > > > > > > This is an interesting idea, but seems likely to get in the way of the > > > original idea of this KIP > > > -- if we always process standby tasks first, then if we are assigned a > > new > > > standby task we > > > will have to wait for it to catch up completely before processing any > > > active tasks! That's > > > even worse than the situation this KIP is trying to help with, since a > > new > > > standby task has to > > > restore from 0 (whereas an active task at least can take over from > > wherever > > > the standby was). > > > > > > During restoration -- while there exist any restoring tasks -- I think > > it's > > > reasonable to de-prioritize the > > > standby tasks and just process restoring and active tasks so both can > > make > > > progress. But we should > > > let them catch up afterwards somehow -- maybe we can apply some kind of > > > heuristic, like "if we haven't > > > processed standbys for X iterations, or Y milliseconds, do so now." > > > > > > Actually, it might even be beneficial to avoid processing standbys a > > record > > > or two at a time and instead > > > wait for a large enough batch to build up for the RocksDB bulk-loading > > > benefits. > > > > > > I think the "use dedicated threads for standby" is the more promising > end > > > goal, especially since > > > if we split restoration into "restoring tasks" then active and standbys > > > share almost nothing. But > > > that seems like follow-up work to the current KIP :) > > > > > > On Thu, Aug 8, 2019 at 5:31 PM Sophie Blee-Goldman < > sop...@confluent.io> > > > wrote: > > > > > > > Stateful tasks with logging disabled seem to be an interesting edge > > case. > > > > On the one hand, > > > > for balancing purposes they should be considered stateful since as > > > > Guozhang pointed out > > > > they are still "heavy" in IO costs. But for "catching up" purposes, > ie > > > > when allocating standby > > > > tasks that will become active tasks, they should be considered > > stateless > > > > as there is so > > > > meaningful sense of their lag. We should never allocate standby tasks > > for > > > > them during the > > > > first rebalance, but should ensure they are evenly distributed across > > > > instances. Maybe we > > > > should split these into a third category -- after we assign all > > stateful > > > > tasks with logging, we > > > > then distribute the set of logging-disabled stateful tasks to improve > > > > balance, before lastly > > > > distributing stateless tasks? > > > > > > > > This actually leads into what I was just thinking, which is that we > > > really > > > > should distinguish the > > > > "catch-up" standbys from normal standbys as well as distinguishing > > > > actively processing tasks > > > > from active tasks that are still in the restore phase. It's somewhat > > > > awkward that today, some > > > > active tasks just start processing immediately while others behave > more > > > > like standby than active > > > > tasks for some time, before switching to real active. They first use > > the > > > > restoreConsumer, then > > > > later only the "normal" consumer. > > > > > > > > However, this restore period is still distinct from normal standbys > in > > a > > > > lot of ways -- the code path > > > > for restoring is different than for updating standbys, for example in > > how > > > > long we block on #poll. > > > > So in addition to giving them their own name -- let's go with > restoring > > > > task for now -- they really > > > > do seem to deserve being their own distinct task. We can optimize > them > > > for > > > > efficient conversion > > > > to active tasks since we know that's what they will be. > > > > > > > > This resolves some of the awkwardness of dealing with the special > case > > > > mentioned above: we > > > > find a balanced assignment of stateful and stateless tasks, and > create > > > > restoring tasks as needed. > > > > If logging is disabled, no restoring task is created. > > > > > > > > > > > > On Thu, Aug 8, 2019 at 3:44 PM Guozhang Wang <wangg...@gmail.com> > > wrote: > > > > > > > >> Regarding 3) above: I think for active task they should still be > > > >> considered > > > >> stateful since the processor would still pay IO cost accessing the > > > store, > > > >> but they would not have standby tasks? > > > >> > > > >> On Thu, Aug 8, 2019 at 7:49 AM Bruno Cadonna <br...@confluent.io> > > > wrote: > > > >> > > > >> > Hi, > > > >> > > > > >> > Thank you for the KIP! > > > >> > > > > >> > Some questions/comments: > > > >> > > > > >> > 1. I am wondering if the "stand-by" tasks that catch up state > before > > > >> > the active task is switched deserve its own name in this KIP and > > maybe > > > >> > in the code. We have already stated that they are not true > stand-by > > > >> > tasks, they are not configured through `num.standby.replicas`, and > > > >> > maybe they have also other properties that distinguish them from > > true > > > >> > stand-by tasks of which we are not aware yet. For example, they > may > > be > > > >> > prioritized differently than other tasks. Furthermore, the name > > > >> > "stand-by" does not really fit with the planned functionality of > > those > > > >> > tasks. In the following, I will call them false stand-by tasks. > > > >> > > > > >> > 2. Did you consider to trigger the probing rebalances not at > regular > > > >> > time intervals but when the false stand-by tasks reach an > acceptable > > > >> > lag? If you did consider, could you add a paragraph why you > rejected > > > >> > this idea to the "Rejected Alternatives" section. > > > >> > > > > >> > 3. Are tasks that solely contain stores with disabled logging > > > >> > classified as stateful or stateless in the algorithm? I would > guess > > > >> > stateless, although if possible they should be assigned to the > same > > > >> > instance they had run before the rebalance. As far as I can see > this > > > >> > special case is not handled in the algorithm. > > > >> > > > > >> > Best, > > > >> > Bruno > > > >> > > > > >> > > > > >> > > > > >> > On Thu, Aug 8, 2019 at 8:24 AM Guozhang Wang <wangg...@gmail.com> > > > >> wrote: > > > >> > > > > > >> > > 1. Sounds good, just wanted to clarify; and it may worth > > documenting > > > >> it > > > >> > so > > > >> > > that users would not be surprised when monitoring their > footprint. > > > >> > > > > > >> > > 2. Hmm I see... I think the trade-off can be described as "how > > much > > > >> > > imbalance would bother you to be willing to pay another > rebalance, > > > >> along > > > >> > > with potentially more restoration lag", and the current > definition > > > of > > > >> > > rebalance_factor can be considered as a rough measurement of > that > > > >> > > imbalance. Of course one can argue that a finer grained > > measurement > > > >> could > > > >> > > be "resource footprint" like CPU / storage of each instance like > > we > > > >> have > > > >> > in > > > >> > > Kafka broker auto balancing tools, but I'd prefer not doing that > > as > > > >> part > > > >> > of > > > >> > > the library but more as an operational tool in the future. On > the > > > >> other > > > >> > > hand, I've seen stateful and stateless tasks having very > different > > > >> load, > > > >> > > and sometimes the only bottleneck of a Streams app is just one > > > >> stateful > > > >> > > sub-topology and whoever gets tasks of that sub-topology become > > > >> hotspot > > > >> > > (and that's why our algorithm tries to balance per sub-topology > as > > > >> well), > > > >> > > so maybe we can just consider stateful tasks when calculating > this > > > >> factor > > > >> > > as a very brute force heuristic? > > > >> > > > > > >> > > 3.a. Thinking about this a bit more, maybe it's better not try > to > > > >> tackle > > > >> > an > > > >> > > unseen enemy just yet, and observe if it really emerges later, > and > > > by > > > >> > then > > > >> > > we may have other ways to not starving the standby tasks, for > > > >> example, by > > > >> > > using dedicate threads for standby tasks or even consider having > > > >> higher > > > >> > > priority for standby than active so that we always try to caught > > up > > > >> > standby > > > >> > > first, then process active; and if active's lagging compared to > > > >> > > log-end-offset is increasing then we should increase capacity, > etc > > > >> etc. > > > >> > > > > > >> > > 4. Actually with KIP-429 this may not be the case: we may not > call > > > >> > > onPartitionsRevoked prior to rebalance any more so would not > > transit > > > >> > state > > > >> > > to PARTITIONS_REVOKED, and hence not cause the state of the > > instance > > > >> to > > > >> > be > > > >> > > REBALANCING. In other words, even if a instance is undergoing a > > > >> rebalance > > > >> > > it's state may still be RUNNING and it may still be processing > > > >> records at > > > >> > > the same time. > > > >> > > > > > >> > > > > > >> > > On Wed, Aug 7, 2019 at 12:14 PM John Roesler <j...@confluent.io > > > > > >> wrote: > > > >> > > > > > >> > > > Hey Guozhang, > > > >> > > > > > > >> > > > Thanks for the review! > > > >> > > > > > > >> > > > 1. Yes, even with `num.standby.replicas := 0`, we will still > > > >> > temporarily > > > >> > > > allocate standby tasks to accomplish a no-downtime task > > migration. > > > >> > > > Although, I'd argue that this doesn't really violate the > config, > > > as > > > >> the > > > >> > > > task isn't a true hot standby. As soon as it catches up, we'll > > > >> > rebalance > > > >> > > > again, that task will become active, and the original instance > > > that > > > >> > hosted > > > >> > > > the active task will no longer have the task assigned at all. > > Once > > > >> the > > > >> > > > stateDirCleaner kicks in, we'll free the disk space from it, > and > > > >> > return to > > > >> > > > the steady-state of having just one copy of the task in the > > > cluster. > > > >> > > > > > > >> > > > We can of course do without this, but I feel the current > > proposal > > > is > > > >> > > > operationally preferable, since it doesn't make configuring > > > >> > hot-standbys a > > > >> > > > pre-requisite for fast rebalances. > > > >> > > > > > > >> > > > 2. Yes, I think your interpretation is what we intended. The > > > default > > > >> > > > balance_factor would be 1, as it is implicitly today. What > this > > > >> does is > > > >> > > > allows operators to trade off less balanced assignments > against > > > >> fewer > > > >> > > > rebalances. If you have lots of space capacity in your > > instances, > > > >> this > > > >> > may > > > >> > > > be a perfectly fine tradeoff, and you may prefer for Streams > not > > > to > > > >> > bother > > > >> > > > streaming GBs of data from the broker in pursuit of perfect > > > balance. > > > >> > Not > > > >> > > > married to this configuration, though. It was inspired by the > > > >> related > > > >> > work > > > >> > > > research we did. > > > >> > > > > > > >> > > > 3. I'll take a look > > > >> > > > > > > >> > > > 3a. I think this is a good idea. I'd classify it as a type of > > grey > > > >> > failure > > > >> > > > detection. It may make more sense to tackle grey failures as > > part > > > of > > > >> > the > > > >> > > > heartbeat protocol (as I POCed here: > > > >> > > > https://github.com/apache/kafka/pull/7096/files). WDYT? > > > >> > > > > > > >> > > > 4. Good catch! I didn't think about that before. Looking at it > > > now, > > > >> > though, > > > >> > > > I wonder if we're actually protected already. The > > stateDirCleaner > > > >> > thread > > > >> > > > only executes if the instance is in RUNNING state, and KIP-441 > > > >> > proposes to > > > >> > > > use "probing rebalances" to report task lag. Hence, during the > > > >> window > > > >> > > > between when the instance reports a lag and the assignor > makes a > > > >> > decision > > > >> > > > about it, the instance should remain in REBALANCING state, > > right? > > > If > > > >> > so, > > > >> > > > then this should prevent the race condition. If not, then we > do > > > >> indeed > > > >> > need > > > >> > > > to do something about it. > > > >> > > > > > > >> > > > 5. Good idea. I think that today, you can only see the > consumer > > > lag, > > > >> > which > > > >> > > > is a poor substitute. I'll add some metrics to the proposal. > > > >> > > > > > > >> > > > Thanks again for the comments! > > > >> > > > -John > > > >> > > > > > > >> > > > On Tue, Aug 6, 2019 at 4:27 PM Guozhang Wang < > > wangg...@gmail.com> > > > >> > wrote: > > > >> > > > > > > >> > > > > Hello Sophie, > > > >> > > > > > > > >> > > > > Thanks for the proposed KIP. I left some comments on the > wiki > > > >> itself, > > > >> > > > and I > > > >> > > > > think I'm still not very clear on a couple or those: > > > >> > > > > > > > >> > > > > 1. With this proposal, does that mean with > > num.standby.replicas > > > == > > > >> > 0, we > > > >> > > > > may sometimes still have some standby tasks which may > violate > > > the > > > >> > config? > > > >> > > > > > > > >> > > > > 2. I think I understand the rationale to consider lags that > is > > > >> below > > > >> > the > > > >> > > > > specified threshold to be equal, rather than still > considering > > > >> 5000 > > > >> > is > > > >> > > > > better than 5001 -- we do not want to "over-optimize" and > > > >> potentially > > > >> > > > falls > > > >> > > > > into endless rebalances back and forth. > > > >> > > > > > > > >> > > > > But I'm not clear about the rationale of the second > parameter > > of > > > >> > > > > > constrainedBalancedAssignment(StatefulTasksToRankedCandidates, > > > >> > > > > balance_factor): > > > >> > > > > > > > >> > > > > Does that mean, e.g. with balance_factor of 3, we'd consider > > two > > > >> > > > > assignments one resulting balance_factor 0 and one resulting > > > >> > > > balance_factor > > > >> > > > > 3 to be equally optimized assignment and therefore may "stop > > > >> early"? > > > >> > This > > > >> > > > > was not very convincing to me :P > > > >> > > > > > > > >> > > > > 3. There are a couple of minor comments about the algorithm > > > >> itself, > > > >> > left > > > >> > > > on > > > >> > > > > the wiki page since it needs to refer to the exact line and > > > better > > > >> > > > > displayed there. > > > >> > > > > > > > >> > > > > 3.a Another wild thought about the threshold itself: today > the > > > >> > assignment > > > >> > > > > itself is memoryless, so we would not know if the reported > > > >> `TaskLag` > > > >> > > > itself > > > >> > > > > is increasing or decreasing even if the current value is > under > > > the > > > >> > > > > threshold. I wonder if it worthy to make it a bit more > > > >> complicated to > > > >> > > > track > > > >> > > > > task lag trend at the assignor? Practically it may not be > very > > > >> > uncommon > > > >> > > > > that stand-by tasks are not keeping up due to the fact that > > > other > > > >> > active > > > >> > > > > tasks hosted on the same thread is starving the standby > tasks. > > > >> > > > > > > > >> > > > > 4. There's a potential race condition risk when reporting > > > >> `TaskLags` > > > >> > in > > > >> > > > the > > > >> > > > > subscription: right after reporting it to the leader, the > > > cleanup > > > >> > thread > > > >> > > > > kicks in and deletes the state directory. If the task was > > > assigned > > > >> > to the > > > >> > > > > host it would cause it to restore from beginning and > > effectively > > > >> > make the > > > >> > > > > seemingly optimized assignment very sub-optimal. > > > >> > > > > > > > >> > > > > To be on the safer side we should consider either prune out > > > those > > > >> > tasks > > > >> > > > > that are "close to be cleaned up" in the subscription, or we > > > >> should > > > >> > delay > > > >> > > > > the cleanup right after we've included them in the > > subscription > > > in > > > >> > case > > > >> > > > > they are been selected as assigned tasks by the assignor. > > > >> > > > > > > > >> > > > > 5. This is a meta comment: I think it would be helpful to > add > > > some > > > >> > user > > > >> > > > > visibility on the standby tasks lagging as well, via metrics > > for > > > >> > example. > > > >> > > > > Today it is hard for us to observe how far are our current > > > standby > > > >> > tasks > > > >> > > > > compared to the active tasks and whether that lag is being > > > >> > increasing or > > > >> > > > > decreasing. As a follow-up task, for example, the rebalance > > > should > > > >> > also > > > >> > > > be > > > >> > > > > triggered if we realize that some standby task's lag is > > > increasing > > > >> > > > > indefinitely means that it cannot keep up (which is another > > > >> indicator > > > >> > > > > either you need to add more resources with the num.standbys > or > > > >> your > > > >> > are > > > >> > > > > still not balanced enough). > > > >> > > > > > > > >> > > > > > > > >> > > > > On Tue, Aug 6, 2019 at 1:32 PM Sophie Blee-Goldman < > > > >> > sop...@confluent.io> > > > >> > > > > wrote: > > > >> > > > > > > > >> > > > > > Hey all, > > > >> > > > > > > > > >> > > > > > I'd like to kick off discussion on KIP-441, aimed at the > > long > > > >> > restore > > > >> > > > > times > > > >> > > > > > in Streams during which further active processing and IQ > are > > > >> > blocked. > > > >> > > > > > Please give it a read and let us know your thoughts > > > >> > > > > > > > > >> > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > >> > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-441:+Smooth+Scaling+Out+for+Kafka+Streams > > > >> > > > > > > > > >> > > > > > Cheers, > > > >> > > > > > Sophie > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > > >> > > > > -- > > > >> > > > > -- Guozhang > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > > >> > > -- > > > >> > > -- Guozhang > > > >> > > > > >> > > > >> > > > >> -- > > > >> -- Guozhang > > > >> > > > > > > > > > > > > > -- > > -- Guozhang > > > -- -- Guozhang
