Hi Mathias, I missed out on the metrics part.
I have added the new metric in the proposed changes section along with the small re-wording that you talked about. Let me know if that makes sense. Thanks! Sagar. On Fri, Sep 10, 2021 at 3:45 AM Matthias J. Sax <mj...@apache.org> wrote: > Thanks for the KIP. > > There was some discussion about adding a metric on the thread, but the > KIP does not contain anything about it. Did we drop this suggestion or > was the KIP not updated accordingly? > > > Nit: > > > This would be a global config applicable per processing topology > > Can we change this to `per Kafka Streams instance.` > > Atm, a Stream instance executes a single topology, so it does not make > any effective difference right now. However, it seems better (more > logical) to bind the config to the instance (not the topology the > instance executes). > > > -Matthias > > On 9/2/21 6:08 AM, Sagar wrote: > > Thanks Guozhang and Luke. > > > > I have updated the KIP with all the suggested changes. > > > > Do you think we could start voting for this? > > > > Thanks! > > Sagar. > > > > On Thu, Sep 2, 2021 at 8:26 AM Luke Chen <show...@gmail.com> wrote: > > > >> Thanks for the KIP. Overall LGTM. > >> > >> Just one thought, if we "rename" the config directly as mentioned in the > >> KIP, would that break existing applications? > >> Should we deprecate the old one first, and make the old/new names > co-exist > >> for some period of time? > >> > >> Public Interfaces > >> > >> - Adding a new config *input.buffer.max.bytes *applicable at a > topology > >> level. The importance of this config would be *Medium*. > >> - Renaming *cache.max.bytes.buffering* to > *statestore.cache.max.bytes*. > >> > >> > >> > >> Thank you. > >> Luke > >> > >> On Thu, Sep 2, 2021 at 1:50 AM Guozhang Wang <wangg...@gmail.com> > wrote: > >> > >>> Currently the state store cache size default value is 10MB today, which > >>> arguably is rather small. So I'm thinking maybe for this config default > >> to > >>> 512MB. > >>> > >>> Other than that, LGTM. > >>> > >>> On Sat, Aug 28, 2021 at 11:34 AM Sagar <sagarmeansoc...@gmail.com> > >> wrote: > >>> > >>>> Thanks Guozhang and Sophie. > >>>> > >>>> Yeah a small default value would lower the throughput. I didn't quite > >>>> realise it earlier. It's slightly hard to predict this value so I > would > >>>> guess around 1/2 GB to 1 GB? WDYT? > >>>> > >>>> Regarding the renaming of the config and the new metric, sure would > >>> include > >>>> it in the KIP. > >>>> > >>>> Lastly, importance would also. be added. I guess Medium should be ok. > >>>> > >>>> Thanks! > >>>> Sagar. > >>>> > >>>> > >>>> On Sat, Aug 28, 2021 at 10:42 AM Sophie Blee-Goldman > >>>> <sop...@confluent.io.invalid> wrote: > >>>> > >>>>> 1) I agree that we should just distribute the bytes evenly, at least > >>> for > >>>>> now. It's simpler to understand and > >>>>> we can always change it later, plus it makes sense to keep this > >> aligned > >>>>> with how the cache works today > >>>>> > >>>>> 2) +1 to being conservative in the generous sense, it's just not > >>>> something > >>>>> we can predict with any degree > >>>>> of accuracy and even if we could, the appropriate value is going to > >>>> differ > >>>>> wildly across applications and use > >>>>> cases. We might want to just pick some multiple of the default cache > >>>> size, > >>>>> and maybe do some research on > >>>>> other relevant defaults or sizes (default JVM heap, size of available > >>>>> memory in common hosts eg EC2 > >>>>> instances, etc). We don't need to worry as much about erring on the > >>> side > >>>> of > >>>>> too big, since other configs like > >>>>> the max.poll.records will help somewhat to keep it from exploding. > >>>>> > >>>>> 4) 100%, I always found the *cache.max.bytes.buffering* config name > >> to > >>> be > >>>>> incredibly confusing. Deprecating this in > >>>>> favor of "*statestore.cache.max.bytes*" and aligning it to the new > >>> input > >>>>> buffer config sounds good to me to include here. > >>>>> > >>>>> 5) The KIP should list all relevant public-facing changes, including > >>>>> metadata like the config's "Importance". Personally > >>>>> I would recommend Medium, or even High if we're really worried about > >>> the > >>>>> default being wrong for a lot of users > >>>>> > >>>>> Thanks for the KIP, besides those few things that Guozhang brought up > >>> and > >>>>> the config importance, everything SGTM > >>>>> > >>>>> -Sophie > >>>>> > >>>>> On Thu, Aug 26, 2021 at 2:41 PM Guozhang Wang <wangg...@gmail.com> > >>>> wrote: > >>>>> > >>>>>> 1) I meant for your proposed solution. I.e. to distribute the > >>>> configured > >>>>>> bytes among threads evenly. > >>>>>> > >>>>>> 2) I was actually thinking about making the default a large enough > >>>> value > >>>>> so > >>>>>> that we would not introduce performance regression: thinking about > >> a > >>>> use > >>>>>> case with many partitions and each record may be large, then > >>>> effectively > >>>>> we > >>>>>> would only start pausing when the total bytes buffered is pretty > >>> large. > >>>>> If > >>>>>> we set the default value to small, we would be "more aggressive" on > >>>>> pausing > >>>>>> which may impact throughput. > >>>>>> > >>>>>> 3) Yes exactly, this would naturally be at the "partition-group" > >>> class > >>>>>> since that represents the task's all input partitions. > >>>>>> > >>>>>> 4) This is just a bold thought, I'm interested to see other's > >>> thoughts. > >>>>>> > >>>>>> > >>>>>> Guozhang > >>>>>> > >>>>>> On Mon, Aug 23, 2021 at 4:10 AM Sagar <sagarmeansoc...@gmail.com> > >>>> wrote: > >>>>>> > >>>>>>> Thanks Guozhang. > >>>>>>> > >>>>>>> 1) Just for my confirmation, when you say we should proceed with > >>> the > >>>>> even > >>>>>>> distribution of bytes, are you referring to the Proposed Solution > >>> in > >>>>> the > >>>>>>> KIP or the option you had considered in the JIRA? > >>>>>>> 2) Default value for the config is something that I missed. I > >> agree > >>>> we > >>>>>>> can't have really large values as it might be detrimental to the > >>>>>>> performance. Maybe, as a starting point, we assume that only 1 > >>> Stream > >>>>>> Task > >>>>>>> is running so what could be the ideal value in such a scenario? > >>>>> Somewhere > >>>>>>> around 10MB similar to the caching config? > >>>>>>> 3) When you say, *a task level metric indicating the current > >>> totally > >>>>>>> aggregated metrics, * you mean the bytes aggregated at a task > >>> level? > >>>>>>> 4) I am ok with the name change, but would like to know others' > >>>>> thoughts. > >>>>>>> > >>>>>>> Thanks! > >>>>>>> Sagar. > >>>>>>> > >>>>>>> On Sun, Aug 22, 2021 at 11:54 PM Guozhang Wang < > >> wangg...@gmail.com > >>>> > >>>>>> wrote: > >>>>>>> > >>>>>>>> Thanks Sagar for writing this PR. > >>>>>>>> > >>>>>>>> I think twice about the options that have been proposed in > >>>>>>>> https://issues.apache.org/jira/browse/KAFKA-13152, and feel > >> that > >>>> at > >>>>>> the > >>>>>>>> moment it's simpler to just do the even distribution of the > >>>>> configured > >>>>>>>> total bytes. My rationale is that right now we have a static > >>> tasks > >>>> -> > >>>>>>>> threads mapping, and hence each partition would only be fetched > >>> by > >>>> a > >>>>>>> single > >>>>>>>> thread / consumer at a given time. If in the future we break > >> that > >>>>>> static > >>>>>>>> mapping into dynamic mapping, then we would not be able to do > >>> this > >>>>> even > >>>>>>>> distribution. Instead we would have other threads polling from > >>>>> consumer > >>>>>>>> only, and those threads would be responsible for checking the > >>>> config > >>>>>> and > >>>>>>>> pause non-empty partitions if it goes beyond the threshold. But > >>>> since > >>>>>> at > >>>>>>>> that time we would not change the config but just how it would > >> be > >>>>>>>> implemented behind the scenes we would not need another KIP to > >>>> change > >>>>>> it. > >>>>>>>> > >>>>>>>> Some more comments: > >>>>>>>> > >>>>>>>> 1. We need to discuss a bit about the default value of this new > >>>>> config. > >>>>>>>> Personally I think we need to be a bit conservative with large > >>>> values > >>>>>> so > >>>>>>>> that it would not have any perf regression compared with old > >>>> configs > >>>>>>>> especially with large topology and large number of partitions. > >>>>>>>> 2. I looked at the existing metrics, and do not have > >>> corresponding > >>>>>>> sensors. > >>>>>>>> How about also adding a task level metric indicating the > >> current > >>>>>> totally > >>>>>>>> aggregated metrics. The reason I do not suggest this metric on > >>> the > >>>>>>>> per-thread level is that in the future we may break the static > >>>>> mapping > >>>>>> of > >>>>>>>> tasks -> threads. > >>>>>>>> > >>>>>>>> [optional] As an orthogonal thought, I'm thinking maybe we can > >>>> rename > >>>>>> the > >>>>>>>> other "*cache.max.bytes.buffering*" as > >>> "statestore.cache.max.bytes" > >>>>>> (via > >>>>>>>> deprecation of course), piggy-backed in this KIP? Would like to > >>>> hear > >>>>>>>> others' thoughts. > >>>>>>>> > >>>>>>>> > >>>>>>>> Guozhang > >>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> On Sun, Aug 22, 2021 at 9:29 AM Sagar < > >> sagarmeansoc...@gmail.com > >>>> > >>>>>> wrote: > >>>>>>>> > >>>>>>>>> Hi All, > >>>>>>>>> > >>>>>>>>> I would like to start a discussion on the following KIP: > >>>>>>>>> > >>>>>>>> > >>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >> > https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186878390 > >>>>>>>>> > >>>>>>>>> Thanks! > >>>>>>>>> Sagar. > >>>>>>>>> > >>>>>>>> > >>>>>>>> > >>>>>>>> -- > >>>>>>>> -- Guozhang > >>>>>>>> > >>>>>>> > >>>>>> > >>>>>> > >>>>>> -- > >>>>>> -- Guozhang > >>>>>> > >>>>> > >>>> > >>> > >>> > >>> -- > >>> -- Guozhang > >>> > >> > > >
