Sorry, I meant `CompletionStage` (instead of CompletableFuture) as this is the interface.
Best, ——— Josep Prat Aiven Deutschland GmbH Immanuelkirchstraße 26, 10405 Berlin Amtsgericht Charlottenburg, HRB 209739 B Geschäftsführer: Oskari Saarenmaa & Hannu Valtonen m: +491715557497 w: aiven.io e: josep.p...@aiven.io On Wed, May 26, 2021, 15:36 Josep Prat <josep.p...@aiven.io> wrote: > Hi, > If I may, I would like to suggest that instead of using Java's `Future` > class on the API's, it would be better to use `CompletableFuture`. This > would offer the possibility of applying callbacks on its completion for > example. > > Best, > > On Wed, May 26, 2021 at 3:28 PM Matthew de Detrich > <matthew.dedetr...@aiven.io.invalid> wrote: > >> Maybe there was a miscommunication but I agree with everything you said, I >> was just clarifying what my definition of blocking is (because I think >> there was a misunderstanding). >> >> And yes you are right, there is a limited amount of threads which is why >> blocking is a bad thing because having threads sitting around waiting/not >> doing anything is a waste of resources but ultimately this is also a >> performance problem because if you don't block you can simply process more >> IO tasks on a given machine/instance (hence greater performance). >> >> In any case, as is clarified the current behavior of send() needs to be >> fixed. It's returning a Future but since it's internally blocking and >> using >> the caller's thread from an API perspective it gives the incorrect >> impression that it's asynchronous (when it's not). >> >> On Wed, May 26, 2021 at 3:15 PM Ryanne Dolan <ryannedo...@gmail.com> >> wrote: >> >> > Matthew, it's more than performance tho. In many frameworks the number >> of >> > request threads is purposefully constrained, and blocking one means you >> > have one less to handle requests with. When you're handling a large >> amount >> > of requests with a small number of threads, any blocking can lead to >> thread >> > exhaustion. >> > >> > For this reason, you'll often see send() wrapped in a future or thread >> > pool. But it's surprising that this would be required, since send() >> already >> > returns a future. >> > >> > Additionally, even when send() does not actually block, it does a lot of >> > work on the caller's thread, which is likewise surprising given a >> future is >> > returned. The effect is the same: less threads are available to handle >> > requests, and you risk thread exhaustion. >> > >> > I think we may incidentally improve performance if we introduce an >> internal >> > thread pool, but the primary motivation here, at least for me, is to fix >> > the lie the API is telling, not to improve performance. >> > >> > Ryanne >> > >> > >> > >> > On Wed, May 26, 2021, 6:51 AM Matthew de Detrich >> > <matthew.dedetr...@aiven.io.invalid> wrote: >> > >> > > I think we may need to clarify terminology here, at least to me >> blocking >> > > means suspending a current thread to wait for some operation (which is >> > > wasteful if we are dealing with IO bound tasks). In other words, the >> > > "blocking" is an implementation detail on how to wait rather than >> whether >> > > we need to wait or not, so to me this is more of a performance >> question. >> > > >> > > In the scenario you describe of kafka clients producing too many >> > messages, >> > > as you said buffering is what should be done but I wouldn't classify >> this >> > > as blocking. >> > > >> > > On Mon, May 24, 2021 at 7:54 PM Colin McCabe <cmcc...@apache.org> >> wrote: >> > > >> > > > Hi all, >> > > > >> > > > I agree that we should give users the option of having a fully async >> > API, >> > > > but I don't think external thread pools or queues are the right >> > direction >> > > > to go here. They add performance overheads and don't address the >> root >> > > > causes of the problem. >> > > > >> > > > There are basically two scenarios where we block, currently. One is >> > when >> > > > we are doing a metadata fetch. I think this is clearly a bug, or at >> > least >> > > > an implementation limitation. From the user's point of view, the >> fact >> > > that >> > > > we are doing a metadata fetch is an implementation detail that >> really >> > > > shouldn't be exposed like this. We have talked about fixing this in >> the >> > > > past. I think we just should spend the time to do it. >> > > > >> > > > The second scenario is where the client has produced too much data >> in >> > too >> > > > little time. This could happen if there is a network glitch, or the >> > > server >> > > > is slower than expected. In this case, the behavior is intentional >> and >> > > not >> > > > a bug. To understand this, think about what would happen if we >> didn't >> > > > block. We would start buffering more and more data in memory, until >> > > finally >> > > > the application died with an out of memory error. That would be >> > > frustrating >> > > > for users and wouldn't add to the usability of Kafka. >> > > > >> > > > We could potentially have an option to handle the out-of-memory >> > scenario >> > > > differently by returning an error code immediately rather than >> > blocking. >> > > > Applications would have to be rewritten to handle this properly, >> but it >> > > is >> > > > a possibility. I suspect that most of them wouldn't use this, but we >> > > could >> > > > offer it as a possibility for async purists (which might include >> > certain >> > > > frameworks). The big problem the users would have to solve is what >> to >> > do >> > > > with the record that they were unable to produce due to the buffer >> full >> > > > issue. >> > > > >> > > > best, >> > > > Colin >> > > > >> > > > >> > > > On Thu, May 20, 2021, at 10:35, Nakamura wrote: >> > > > > > >> > > > > > My suggestion was just do this in multiple steps/phases, firstly >> > > let's >> > > > fix >> > > > > > the issue of send being misleadingly asynchronous (i.e. >> internally >> > > its >> > > > > > blocking) and then later one we can make the various >> > > > > > threadpools configurable with a sane default. >> > > > > >> > > > > I like that approach. I updated the "Which thread should be >> > responsible >> > > > for >> > > > > waiting" part of KIP-739 to add your suggestion as my recommended >> > > > approach, >> > > > > thank you! If no one else has major concerns about that approach, >> > I'll >> > > > > move the alternatives to "rejected alternatives". >> > > > > >> > > > > On Thu, May 20, 2021 at 7:26 AM Matthew de Detrich >> > > > > <matthew.dedetr...@aiven.io.invalid> wrote: >> > > > > >> > > > > > @ >> > > > > > >> > > > > > Nakamura >> > > > > > On Wed, May 19, 2021 at 7:35 PM Nakamura <nny...@gmail.com> >> wrote: >> > > > > > >> > > > > > > @Ryanne: >> > > > > > > In my mind's eye I slightly prefer the throwing the "cannot >> > > enqueue" >> > > > > > > exception to satisfying the future immediately with the >> "cannot >> > > > enqueue" >> > > > > > > exception? But I agree, it would be worth doing more >> research. >> > > > > > > >> > > > > > > @Matthew: >> > > > > > > >> > > > > > > > 3. Using multiple thread pools is definitely recommended for >> > > > different >> > > > > > > > types of tasks, for serialization which is CPU bound you >> > > definitely >> > > > > > would >> > > > > > > > want to use a bounded thread pool that is fixed by the >> number >> > of >> > > > CPU's >> > > > > > > (or >> > > > > > > > something along those lines). >> > > > > > > > >> > > https://gist.github.com/djspiewak/46b543800958cf61af6efa8e072bfd5c >> > > > is >> > > > > > a >> > > > > > > > very good guide on this topic >> > > > > > > I think this guide is good in general, but I would be >> hesitant to >> > > > follow >> > > > > > > its guidance re: offloading serialization without benchmarking >> > it. >> > > > My >> > > > > > > understanding is that context-switches have gotten much >> cheaper, >> > > and >> > > > that >> > > > > > > gains from cache locality are small, but they're not nothing. >> > > > Especially >> > > > > > > if the workload has a very small serialization cost, I >> wouldn't >> > be >> > > > > > shocked >> > > > > > > if it made it slower. I feel pretty strongly that we should >> do >> > > more >> > > > > > > research here before unconditionally encouraging serialization >> > in a >> > > > > > > threadpool. If people think it's important to do it here (eg >> if >> > we >> > > > think >> > > > > > > it would mean another big API change) then we should start >> > thinking >> > > > about >> > > > > > > what benchmarking we can do to gain higher confidence in this >> > kind >> > > of >> > > > > > > change. However, I don't think it would change semantics as >> > > > > > substantially >> > > > > > > as we're proposing here, so I would vote for pushing this to a >> > > > subsequent >> > > > > > > KIP. >> > > > > > > >> > > > > > Of course, its all down to benchmarking, benchmarking and >> > > benchmarking. >> > > > > > Ideally speaking you want to use all of the resources available >> to >> > > > you, so >> > > > > > if you have a bottleneck in serialization and you have many >> cores >> > > free >> > > > then >> > > > > > using multiple cores may be more appropriate than a single core. >> > > > Typically >> > > > > > I would expect that using a single thread to do serialization is >> > > > likely to >> > > > > > be the most situation, I was just responding to an earlier point >> > that >> > > > was >> > > > > > made in regards to using ThreadPools for serialization (note >> that >> > you >> > > > can >> > > > > > also just use a ThreadPool that is pinned to a single thread) >> > > > > > >> > > > > > >> > > > > > >> > > > > > > >> > > > > > > > 4. Regarding providing the ability for users to supply their >> > own >> > > > custom >> > > > > > > > ThreadPool this is more of an ergonomics question for the >> API. >> > > > > > Especially >> > > > > > > > when it gets to monitoring/tracing, giving the ability for >> > users >> > > to >> > > > > > > provide >> > > > > > > > their own custom IO/CPU ThreadPools is ideal however as >> stated >> > > > doing so >> > > > > > > > means a lot of boilerplatery changes to the API. Typically >> > > > speaking a >> > > > > > lot >> > > > > > > > of monitoring/tracing/diagnosing is done on >> > > > > > ExecutionContext/ThreadPools >> > > > > > > > (at least on a more rudimentary level) and hence allowing >> users >> > > to >> > > > > > supply >> > > > > > > a >> > > > > > > > global singleton ThreadPool for IO tasks and another for CPU >> > > tasks >> > > > > > makes >> > > > > > > > their lives a lot easier. However due to the large amount of >> > > > changes to >> > > > > > > the >> > > > > > > > API, it may be more appropriate to just use internal thread >> > pools >> > > > (for >> > > > > > > now) >> > > > > > > > since at least it's not any worse than what exists currently >> > and >> > > > this >> > > > > > can >> > > > > > > > be an improvement that is done later? >> > > > > > > Is there an existing threadpool that you suggest we reuse? Or >> > are >> > > > you >> > > > > > > imagining that we make our own internal threadpool, and then >> > maybe >> > > > expose >> > > > > > > configuration flags to manipulate it? For what it's worth, I >> > like >> > > > having >> > > > > > > an internal threadpool (perhaps just FJP.commonpool) and then >> > > > providing >> > > > > > an >> > > > > > > alternative to pass your own threadpool. That way people who >> > want >> > > > finer >> > > > > > > control can get it, and everyone else can do OK with the >> default. >> > > > > > > >> > > > > > Indeed that is what I am saying. The most ideal situation is >> that >> > > > there is >> > > > > > a default internal threadpool that Kafka uses, however users of >> > Kafka >> > > > can >> > > > > > configure there own threadpool. Having a singleton ThreadPool >> for >> > > > blocking >> > > > > > IO, non blocking IO and CPU bound tasks which can be plugged in >> all >> > > of >> > > > your >> > > > > > libraries (including Kafka) makes resource management much >> easier >> > to >> > > > do and >> > > > > > also gives control of users to override specific threadpools for >> > > > > > exceptional cases (i.e. providing a Threadpool that is pinned >> to a >> > > > single >> > > > > > core which tends to give the best latency results if this is >> > > something >> > > > that >> > > > > > is critical for you). >> > > > > > >> > > > > > My suggestion was just do this in multiple steps/phases, firstly >> > > let's >> > > > fix >> > > > > > the issue of send being misleadingly asynchronous (i.e. >> internally >> > > its >> > > > > > blocking) and then later one we can make the various >> > > > > > threadpools configurable with a sane default. >> > > > > > >> > > > > > > >> > > > > > > >> > > > > > > On Wed, May 19, 2021 at 6:01 AM Matthew de Detrich >> > > > > > > <matthew.dedetr...@aiven.io.invalid> wrote: >> > > > > > > >> > > > > > > > Here are my two cents here (note that I am only seeing this >> on >> > a >> > > > > > surface >> > > > > > > > level) >> > > > > > > > >> > > > > > > > 1. If we are going this road it makes sense to do this >> > "properly" >> > > > (i.e. >> > > > > > > > using queues as Ryan suggested). The reason I am saying >> this >> > is >> > > > that >> > > > > > it >> > > > > > > > seems that the original goal of the KIP is for it to be >> used in >> > > > other >> > > > > > > > asynchronous systems and from my personal experience, you >> > really >> > > do >> > > > > > need >> > > > > > > to >> > > > > > > > make the implementation properly asynchronous otherwise it's >> > > > really not >> > > > > > > > that useful. >> > > > > > > > 2. Due to the previous point and what was said by others, >> this >> > is >> > > > > > likely >> > > > > > > > going to break some existing semantics (i.e. people are >> > currently >> > > > > > relying >> > > > > > > > on blocking semantics) so adding another method's/interface >> > plus >> > > > > > > > deprecating the older one is more annoying but ideal. >> > > > > > > > 3. Using multiple thread pools is definitely recommended for >> > > > different >> > > > > > > > types of tasks, for serialization which is CPU bound you >> > > definitely >> > > > > > would >> > > > > > > > want to use a bounded thread pool that is fixed by the >> number >> > of >> > > > CPU's >> > > > > > > (or >> > > > > > > > something along those lines). >> > > > > > > > >> > > https://gist.github.com/djspiewak/46b543800958cf61af6efa8e072bfd5c >> > > > is >> > > > > > a >> > > > > > > > very good guide on this topic >> > > > > > > > 4. Regarding providing the ability for users to supply their >> > own >> > > > custom >> > > > > > > > ThreadPool this is more of an ergonomics question for the >> API. >> > > > > > Especially >> > > > > > > > when it gets to monitoring/tracing, giving the ability for >> > users >> > > to >> > > > > > > provide >> > > > > > > > their own custom IO/CPU ThreadPools is ideal however as >> stated >> > > > doing so >> > > > > > > > means a lot of boilerplatery changes to the API. Typically >> > > > speaking a >> > > > > > lot >> > > > > > > > of monitoring/tracing/diagnosing is done on >> > > > > > ExecutionContext/ThreadPools >> > > > > > > > (at least on a more rudimentary level) and hence allowing >> users >> > > to >> > > > > > > supply a >> > > > > > > > global singleton ThreadPool for IO tasks and another for CPU >> > > tasks >> > > > > > makes >> > > > > > > > their lives a lot easier. However due to the large amount of >> > > > changes to >> > > > > > > the >> > > > > > > > API, it may be more appropriate to just use internal thread >> > pools >> > > > (for >> > > > > > > now) >> > > > > > > > since at least it's not any worse than what exists currently >> > and >> > > > this >> > > > > > can >> > > > > > > > be an improvement that is done later? >> > > > > > > > >> > > > > > > > On Wed, May 19, 2021 at 2:56 AM Ryanne Dolan < >> > > > ryannedo...@gmail.com> >> > > > > > > > wrote: >> > > > > > > > >> > > > > > > > > I was thinking the sender would typically wrap send() in a >> > > > > > > backoff/retry >> > > > > > > > > loop, or else ignore any failures and drop sends on the >> floor >> > > > > > > > > (fire-and-forget), and in both cases I think failing >> > > immediately >> > > > is >> > > > > > > > better >> > > > > > > > > than blocking for a new spot in the queue or >> asynchronously >> > > > failing >> > > > > > > > > somehow. >> > > > > > > > > >> > > > > > > > > I think a failed future is adequate for the "explicit >> > > > backpressure >> > > > > > > > signal" >> > > > > > > > > while avoiding any blocking anywhere. I think if we try to >> > > > > > > asynchronously >> > > > > > > > > signal the caller of failure (either by asynchronously >> > failing >> > > > the >> > > > > > > future >> > > > > > > > > or invoking a callback off-thread or something) we'd force >> > the >> > > > caller >> > > > > > > to >> > > > > > > > > either block or poll waiting for that signal, which >> somewhat >> > > > defeats >> > > > > > > the >> > > > > > > > > purpose we're after. And of course blocking for a spot in >> the >> > > > queue >> > > > > > > > > definitely defeats the purpose (tho perhaps ameliorates >> the >> > > > problem >> > > > > > > > some). >> > > > > > > > > >> > > > > > > > > Throwing an exception to the caller directly (not via the >> > > > future) is >> > > > > > > > > another option with precedent in Kafka clients, tho it >> > doesn't >> > > > seem >> > > > > > as >> > > > > > > > > ergonomic to me. >> > > > > > > > > >> > > > > > > > > It would be interesting to analyze some existing usage and >> > > > determine >> > > > > > > how >> > > > > > > > > difficult it would be to convert it to the various >> proposed >> > > APIs. >> > > > > > > > > >> > > > > > > > > Ryanne >> > > > > > > > > >> > > > > > > > > On Tue, May 18, 2021, 3:27 PM Nakamura <nny...@gmail.com> >> > > wrote: >> > > > > > > > > >> > > > > > > > > > Hi Ryanne, >> > > > > > > > > > >> > > > > > > > > > Hmm, that's an interesting idea. Basically it would >> mean >> > > that >> > > > > > after >> > > > > > > > > > calling send, you would also have to check whether the >> > > returned >> > > > > > > future >> > > > > > > > > had >> > > > > > > > > > failed with a specific exception. I would be open to >> it, >> > > > although >> > > > > > I >> > > > > > > > > think >> > > > > > > > > > it might be slightly more surprising, since right now >> the >> > > > paradigm >> > > > > > is >> > > > > > > > > > "enqueue synchronously, the future represents whether we >> > > > succeeded >> > > > > > in >> > > > > > > > > > sending or not" and the new one would be "enqueue >> > > > synchronously, >> > > > > > the >> > > > > > > > > future >> > > > > > > > > > either represents whether we succeeded in enqueueing or >> not >> > > (in >> > > > > > which >> > > > > > > > > case >> > > > > > > > > > it will be failed immediately if it failed to enqueue) >> or >> > > > whether >> > > > > > we >> > > > > > > > > > succeeded in sending or not". >> > > > > > > > > > >> > > > > > > > > > But you're right, it should be on the table, thank you >> for >> > > > > > suggesting >> > > > > > > > it! >> > > > > > > > > > >> > > > > > > > > > Best, >> > > > > > > > > > Moses >> > > > > > > > > > >> > > > > > > > > > On Tue, May 18, 2021 at 12:23 PM Ryanne Dolan < >> > > > > > ryannedo...@gmail.com >> > > > > > > > >> > > > > > > > > > wrote: >> > > > > > > > > > >> > > > > > > > > > > Moses, in the case of a full queue, could we just >> return >> > a >> > > > failed >> > > > > > > > > future >> > > > > > > > > > > immediately? >> > > > > > > > > > > >> > > > > > > > > > > Ryanne >> > > > > > > > > > > >> > > > > > > > > > > On Tue, May 18, 2021, 10:39 AM Nakamura < >> > nny...@gmail.com> >> > > > > > wrote: >> > > > > > > > > > > >> > > > > > > > > > > > Hi Alexandre, >> > > > > > > > > > > > >> > > > > > > > > > > > Thanks for bringing this up, I think I could use >> some >> > > > feedback >> > > > > > in >> > > > > > > > > this >> > > > > > > > > > > > area. There are two mechanisms here, one for >> slowing >> > > down >> > > > when >> > > > > > > we >> > > > > > > > > > don't >> > > > > > > > > > > > have the relevant metadata, and the other for >> slowing >> > > down >> > > > > > when a >> > > > > > > > > queue >> > > > > > > > > > > has >> > > > > > > > > > > > filled up. Although the first one applies >> backpressure >> > > > > > somewhat >> > > > > > > > > > > > inadvertently, we could still get in trouble if >> we're >> > not >> > > > > > > providing >> > > > > > > > > > > > information to the mechanism that monitors whether >> > we're >> > > > > > queueing >> > > > > > > > too >> > > > > > > > > > > > much. As for the second one, that is a classic >> > > > backpressure >> > > > > > use >> > > > > > > > > case, >> > > > > > > > > > so >> > > > > > > > > > > > it's definitely important that we don't drop that >> > > ability. >> > > > > > > > > > > > >> > > > > > > > > > > > Right now backpressure is applied by blocking, which >> > is a >> > > > > > natural >> > > > > > > > way >> > > > > > > > > > to >> > > > > > > > > > > > apply backpressure in synchronous systems, but can >> lead >> > > to >> > > > > > > > > unnecessary >> > > > > > > > > > > > slowdowns in asynchronous systems. In my opinion, >> the >> > > > safest >> > > > > > way >> > > > > > > > to >> > > > > > > > > > > apply >> > > > > > > > > > > > backpressure in an asynchronous model is to have an >> > > > explicit >> > > > > > > > > > backpressure >> > > > > > > > > > > > signal. A good example would be returning an >> > exception, >> > > > and >> > > > > > > > > providing >> > > > > > > > > > an >> > > > > > > > > > > > optional hook to add a callback onto so that you >> can be >> > > > > > notified >> > > > > > > > when >> > > > > > > > > > > it's >> > > > > > > > > > > > ready to accept more messages. >> > > > > > > > > > > > >> > > > > > > > > > > > However, this would be a really big change to how >> users >> > > use >> > > > > > > > > > > > KafkaProducer#send, so I don't know how much >> appetite >> > we >> > > > have >> > > > > > for >> > > > > > > > > > making >> > > > > > > > > > > > that kind of change. Maybe it would be simpler to >> > remove >> > > > the >> > > > > > > > "don't >> > > > > > > > > > > block >> > > > > > > > > > > > when the per-topic queue is full" from the scope of >> > this >> > > > KIP, >> > > > > > and >> > > > > > > > > only >> > > > > > > > > > > > focus on when metadata is available? The downside >> is >> > > that >> > > > we >> > > > > > > will >> > > > > > > > > > > probably >> > > > > > > > > > > > want to change the API again later to fix this, so >> it >> > > > might be >> > > > > > > > better >> > > > > > > > > > to >> > > > > > > > > > > > just rip the bandaid off now. >> > > > > > > > > > > > >> > > > > > > > > > > > One slightly nasty thing here is that because >> queueing >> > > > order is >> > > > > > > > > > > important, >> > > > > > > > > > > > if we want to use exceptions, we will want to be >> able >> > to >> > > > signal >> > > > > > > the >> > > > > > > > > > > failure >> > > > > > > > > > > > to enqueue to the caller in such a way that they can >> > > still >> > > > > > > enforce >> > > > > > > > > > > message >> > > > > > > > > > > > order if they want. So we can't embed the failure >> > > > directly in >> > > > > > > the >> > > > > > > > > > > returned >> > > > > > > > > > > > future, we should either return two futures >> (nested, or >> > > as >> > > > a >> > > > > > > tuple) >> > > > > > > > > or >> > > > > > > > > > > else >> > > > > > > > > > > > throw an exception to explain a backpressure. >> > > > > > > > > > > > >> > > > > > > > > > > > So there are a few things we should work out here: >> > > > > > > > > > > > >> > > > > > > > > > > > 1. Should we keep the "too many bytes enqueued" >> part of >> > > > this in >> > > > > > > > > scope? >> > > > > > > > > > > (I >> > > > > > > > > > > > would say yes, so that we can minimize churn in this >> > API) >> > > > > > > > > > > > 2. How should we signal backpressure so that it's >> > > > appropriate >> > > > > > for >> > > > > > > > > > > > asynchronous systems? (I would say that we should >> > throw >> > > an >> > > > > > > > > exception. >> > > > > > > > > > > If >> > > > > > > > > > > > we choose this and we want to pursue the queueing >> path, >> > > we >> > > > > > would >> > > > > > > > > *not* >> > > > > > > > > > > want >> > > > > > > > > > > > to enqueue messages that would push us over the >> limit, >> > > and >> > > > > > would >> > > > > > > > only >> > > > > > > > > > > want >> > > > > > > > > > > > to enqueue messages when we're waiting for metadata, >> > and >> > > we >> > > > > > would >> > > > > > > > > want >> > > > > > > > > > to >> > > > > > > > > > > > keep track of the total number of bytes for those >> > > > messages). >> > > > > > > > > > > > >> > > > > > > > > > > > What do you think? >> > > > > > > > > > > > >> > > > > > > > > > > > Best, >> > > > > > > > > > > > Moses >> > > > > > > > > > > > >> > > > > > > > > > > > On Sun, May 16, 2021 at 6:16 AM Alexandre Dupriez < >> > > > > > > > > > > > alexandre.dupr...@gmail.com> wrote: >> > > > > > > > > > > > >> > > > > > > > > > > > > Hello Nakamura, >> > > > > > > > > > > > > >> > > > > > > > > > > > > Thanks for proposing this change. I can see how >> the >> > > > blocking >> > > > > > > > > > behaviour >> > > > > > > > > > > > > can be a problem when integrating with reactive >> > > > frameworks >> > > > > > such >> > > > > > > > as >> > > > > > > > > > > > > Akka. One of the questions I would have is how you >> > > would >> > > > > > handle >> > > > > > > > > back >> > > > > > > > > > > > > pressure and avoid memory exhaustion when the >> > > producer's >> > > > > > buffer >> > > > > > > > is >> > > > > > > > > > > > > full and tasks would start to accumulate in the >> > > > out-of-band >> > > > > > > queue >> > > > > > > > > or >> > > > > > > > > > > > > thread pool introduced with this KIP. >> > > > > > > > > > > > > >> > > > > > > > > > > > > Thanks, >> > > > > > > > > > > > > Alexandre >> > > > > > > > > > > > > >> > > > > > > > > > > > > Le ven. 14 mai 2021 à 15:55, Ryanne Dolan < >> > > > > > > ryannedo...@gmail.com >> > > > > > > > > >> > > > > > > > > a >> > > > > > > > > > > > écrit >> > > > > > > > > > > > > : >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > Makes sense! >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > Ryanne >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > On Fri, May 14, 2021, 9:39 AM Nakamura < >> > > > nny...@gmail.com> >> > > > > > > > wrote: >> > > > > > > > > > > > > > >> > > > > > > > > > > > > > > Hey Ryanne, >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > I see what you're saying about serde blocking, >> > but >> > > I >> > > > > > think >> > > > > > > we >> > > > > > > > > > > should >> > > > > > > > > > > > > > > consider it out of scope for this patch. >> Right >> > now >> > > > we've >> > > > > > > > > nailed >> > > > > > > > > > > > down a >> > > > > > > > > > > > > > > couple of use cases where we can unambiguously >> > say, >> > > > "I >> > > > > > can >> > > > > > > > make >> > > > > > > > > > > > > progress >> > > > > > > > > > > > > > > now" or "I cannot make progress now", which >> makes >> > > it >> > > > > > > possible >> > > > > > > > > to >> > > > > > > > > > > > > offload to >> > > > > > > > > > > > > > > a different thread only if we are unable to >> make >> > > > > > progress. >> > > > > > > > > > > Extending >> > > > > > > > > > > > > this >> > > > > > > > > > > > > > > to CPU work like serde would mean always >> > > offloading, >> > > > > > which >> > > > > > > > > would >> > > > > > > > > > > be a >> > > > > > > > > > > > > > > really big performance change. It might be >> worth >> > > > > > exploring >> > > > > > > > > > anyway, >> > > > > > > > > > > > > but I'd >> > > > > > > > > > > > > > > rather keep this patch focused on improving >> > > > ergonomics, >> > > > > > > > rather >> > > > > > > > > > than >> > > > > > > > > > > > > > > muddying the waters with evaluating >> performance >> > > very >> > > > > > > deeply. >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > I think if we really do want to support serde >> or >> > > > > > > interceptors >> > > > > > > > > > that >> > > > > > > > > > > do >> > > > > > > > > > > > > IO on >> > > > > > > > > > > > > > > the send path (which seems like an >> anti-pattern >> > to >> > > > me), >> > > > > > we >> > > > > > > > > should >> > > > > > > > > > > > > consider >> > > > > > > > > > > > > > > making that a separate SIP, and probably also >> > > > consider >> > > > > > > > changing >> > > > > > > > > > the >> > > > > > > > > > > > > API to >> > > > > > > > > > > > > > > use Futures (or CompletionStages). But I >> would >> > > > rather >> > > > > > > avoid >> > > > > > > > > > scope >> > > > > > > > > > > > > creep, >> > > > > > > > > > > > > > > so that we have a better chance of fixing this >> > part >> > > > of >> > > > > > the >> > > > > > > > > > problem. >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > Yes, I think some exceptions will move to >> being >> > > async >> > > > > > > instead >> > > > > > > > > of >> > > > > > > > > > > > sync. >> > > > > > > > > > > > > > > They'll still be surfaced in the Future, so >> I'm >> > not >> > > > so >> > > > > > > > > confident >> > > > > > > > > > > that >> > > > > > > > > > > > > it >> > > > > > > > > > > > > > > would be that big a shock to users though. >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > Best, >> > > > > > > > > > > > > > > Moses >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > On Thu, May 13, 2021 at 7:44 PM Ryanne Dolan < >> > > > > > > > > > > ryannedo...@gmail.com> >> > > > > > > > > > > > > > > wrote: >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > re serialization, my concern is that >> > > serialization >> > > > > > often >> > > > > > > > > > accounts >> > > > > > > > > > > > > for a >> > > > > > > > > > > > > > > lot >> > > > > > > > > > > > > > > > of the cycles spent before returning the >> > future. >> > > > It's >> > > > > > not >> > > > > > > > > > > blocking >> > > > > > > > > > > > > per >> > > > > > > > > > > > > > > se, >> > > > > > > > > > > > > > > > but it's the same effect from the caller's >> > > > perspective. >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > Moreover, serde impls often block >> themselves, >> > > e.g. >> > > > when >> > > > > > > > > > fetching >> > > > > > > > > > > > > schemas >> > > > > > > > > > > > > > > > from a registry. I suppose it's also >> possible >> > to >> > > > block >> > > > > > in >> > > > > > > > > > > > > Interceptors >> > > > > > > > > > > > > > > > (e.g. writing audit events or metrics), >> which >> > > > happens >> > > > > > > > before >> > > > > > > > > > > serdes >> > > > > > > > > > > > > iiuc. >> > > > > > > > > > > > > > > > So any blocking in either of those plugins >> > would >> > > > block >> > > > > > > the >> > > > > > > > > send >> > > > > > > > > > > > > unless we >> > > > > > > > > > > > > > > > queue first. >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > So I think we want to queue first and do >> > > everything >> > > > > > > > > off-thread >> > > > > > > > > > > when >> > > > > > > > > > > > > using >> > > > > > > > > > > > > > > > the new API, whatever that looks like. I >> just >> > > want >> > > > to >> > > > > > > make >> > > > > > > > > sure >> > > > > > > > > > > we >> > > > > > > > > > > > > don't >> > > > > > > > > > > > > > > do >> > > > > > > > > > > > > > > > that for clients that wouldn't expect it. >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > Another consideration is exception >> handling. If >> > > we >> > > > > > queue >> > > > > > > > > right >> > > > > > > > > > > > away, >> > > > > > > > > > > > > > > we'll >> > > > > > > > > > > > > > > > defer some exceptions that currently are >> thrown >> > > to >> > > > the >> > > > > > > > caller >> > > > > > > > > > > > > (before the >> > > > > > > > > > > > > > > > future is returned). In the new API, the >> send() >> > > > > > wouldn't >> > > > > > > > > throw >> > > > > > > > > > > any >> > > > > > > > > > > > > > > > exceptions, and instead the future would >> fail. >> > I >> > > > think >> > > > > > > that >> > > > > > > > > > might >> > > > > > > > > > > > > mean >> > > > > > > > > > > > > > > that >> > > > > > > > > > > > > > > > a new method signature is required. >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > Ryanne >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > On Thu, May 13, 2021, 2:57 PM Nakamura < >> > > > > > > > > > nakamura.mo...@gmail.com >> > > > > > > > > > > > >> > > > > > > > > > > > > wrote: >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > Hey Ryanne, >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > I agree we should add an additional >> > constructor >> > > > (or >> > > > > > > else >> > > > > > > > an >> > > > > > > > > > > > > additional >> > > > > > > > > > > > > > > > > overload in KafkaProducer#send, but the >> new >> > > > > > constructor >> > > > > > > > > would >> > > > > > > > > > > be >> > > > > > > > > > > > > easier >> > > > > > > > > > > > > > > > to >> > > > > > > > > > > > > > > > > understand) if we're targeting the "user >> > > > provides the >> > > > > > > > > thread" >> > > > > > > > > > > > > approach. >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > From looking at the code, I think we can >> keep >> > > > record >> > > > > > > > > > > > serialization >> > > > > > > > > > > > > on >> > > > > > > > > > > > > > > the >> > > > > > > > > > > > > > > > > user thread, if we consider that an >> important >> > > > part of >> > > > > > > the >> > > > > > > > > > > > > semantics of >> > > > > > > > > > > > > > > > this >> > > > > > > > > > > > > > > > > method. It doesn't seem like >> serialization >> > > > depends >> > > > > > on >> > > > > > > > > > knowing >> > > > > > > > > > > > the >> > > > > > > > > > > > > > > > cluster, >> > > > > > > > > > > > > > > > > I think it's incidental that it comes >> after >> > the >> > > > first >> > > > > > > > > > > "blocking" >> > > > > > > > > > > > > > > segment >> > > > > > > > > > > > > > > > in >> > > > > > > > > > > > > > > > > the method. >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > Best, >> > > > > > > > > > > > > > > > > Moses >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > On Thu, May 13, 2021 at 2:38 PM Ryanne >> Dolan >> > < >> > > > > > > > > > > > > ryannedo...@gmail.com> >> > > > > > > > > > > > > > > > > wrote: >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > Hey Moses, I like the direction here. My >> > > > thinking >> > > > > > is >> > > > > > > > > that a >> > > > > > > > > > > > > single >> > > > > > > > > > > > > > > > > > additional work queue, s.t. send() can >> > > enqueue >> > > > and >> > > > > > > > > return, >> > > > > > > > > > > > seems >> > > > > > > > > > > > > like >> > > > > > > > > > > > > > > > the >> > > > > > > > > > > > > > > > > > lightest touch. However, I don't think >> we >> > can >> > > > > > > trivially >> > > > > > > > > > > process >> > > > > > > > > > > > > that >> > > > > > > > > > > > > > > > > queue >> > > > > > > > > > > > > > > > > > in an internal thread pool without >> subtly >> > > > changing >> > > > > > > > > behavior >> > > > > > > > > > > for >> > > > > > > > > > > > > some >> > > > > > > > > > > > > > > > > users. >> > > > > > > > > > > > > > > > > > For example, users will often run >> send() in >> > > > > > multiple >> > > > > > > > > > threads >> > > > > > > > > > > in >> > > > > > > > > > > > > order >> > > > > > > > > > > > > > > > to >> > > > > > > > > > > > > > > > > > serialize faster, but that wouldn't work >> > > quite >> > > > the >> > > > > > > same >> > > > > > > > > if >> > > > > > > > > > > > there >> > > > > > > > > > > > > were >> > > > > > > > > > > > > > > > an >> > > > > > > > > > > > > > > > > > internal thread pool. >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > For this reason I'm thinking we need to >> > make >> > > > sure >> > > > > > any >> > > > > > > > > such >> > > > > > > > > > > > > changes >> > > > > > > > > > > > > > > are >> > > > > > > > > > > > > > > > > > opt-in. Maybe a new constructor with an >> > > > additional >> > > > > > > > > > > > ThreadFactory >> > > > > > > > > > > > > > > > > parameter. >> > > > > > > > > > > > > > > > > > That would at least clearly indicate >> that >> > > work >> > > > will >> > > > > > > > > happen >> > > > > > > > > > > > > > > off-thread, >> > > > > > > > > > > > > > > > > and >> > > > > > > > > > > > > > > > > > would require opt-in for the new >> behavior. >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > Under the hood, this ThreadFactory >> could be >> > > > used to >> > > > > > > > > create >> > > > > > > > > > > the >> > > > > > > > > > > > > worker >> > > > > > > > > > > > > > > > > > thread that process queued sends, which >> > could >> > > > > > fan-out >> > > > > > > > to >> > > > > > > > > > > > > > > per-partition >> > > > > > > > > > > > > > > > > > threads from there. >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > So then you'd have two ways to send: the >> > > > existing >> > > > > > > way, >> > > > > > > > > > where >> > > > > > > > > > > > > serde >> > > > > > > > > > > > > > > and >> > > > > > > > > > > > > > > > > > interceptors and whatnot are executed on >> > the >> > > > > > calling >> > > > > > > > > > thread, >> > > > > > > > > > > > and >> > > > > > > > > > > > > the >> > > > > > > > > > > > > > > > new >> > > > > > > > > > > > > > > > > > way, which returns right away and uses >> an >> > > > internal >> > > > > > > > > > Executor. >> > > > > > > > > > > As >> > > > > > > > > > > > > you >> > > > > > > > > > > > > > > > point >> > > > > > > > > > > > > > > > > > out, the semantics would be identical in >> > > either >> > > > > > case, >> > > > > > > > and >> > > > > > > > > > it >> > > > > > > > > > > > > would be >> > > > > > > > > > > > > > > > > very >> > > > > > > > > > > > > > > > > > easy for clients to switch. >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > Ryanne >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > On Thu, May 13, 2021, 9:00 AM Nakamura < >> > > > > > > > nny...@gmail.com >> > > > > > > > > > >> > > > > > > > > > > > wrote: >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > Hey Folks, >> > > > > > > > > > > > > > > > > > > I just posted a new proposal >> > > > > > > > > > > > > > > > > > > < >> > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > >> > > >> > >> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446 >> > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > in the wiki. I think we have an >> > > opportunity >> > > > to >> > > > > > > > improve >> > > > > > > > > > the >> > > > > > > > > > > > > > > > > > > KafkaProducer#send user experience. >> It >> > > would >> > > > > > > > certainly >> > > > > > > > > > > make >> > > > > > > > > > > > > our >> > > > > > > > > > > > > > > > lives >> > > > > > > > > > > > > > > > > > > easier. Please take a look! There >> are >> > two >> > > > > > > > subproblems >> > > > > > > > > > > that >> > > > > > > > > > > > I >> > > > > > > > > > > > > > > could >> > > > > > > > > > > > > > > > > use >> > > > > > > > > > > > > > > > > > > guidance on, so I would appreciate >> > feedback >> > > > on >> > > > > > both >> > > > > > > > of >> > > > > > > > > > > them. >> > > > > > > > > > > > > > > > > > > Best, >> > > > > > > > > > > > > > > > > > > Moses >> > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > -- >> > > > > > > > >> > > > > > > > Matthew de Detrich >> > > > > > > > >> > > > > > > > *Aiven Deutschland GmbH* >> > > > > > > > >> > > > > > > > Immanuelkirchstraße 26, 10405 Berlin >> > > > > > > > >> > > > > > > > Amtsgericht Charlottenburg, HRB 209739 B >> > > > > > > > >> > > > > > > > *m:* +491603708037 >> > > > > > > > >> > > > > > > > *w:* aiven.io *e:* matthew.dedetr...@aiven.io >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > > >> > > > > > -- >> > > > > > >> > > > > > Matthew de Detrich >> > > > > > >> > > > > > *Aiven Deutschland GmbH* >> > > > > > >> > > > > > Immanuelkirchstraße 26, 10405 Berlin >> > > > > > >> > > > > > Amtsgericht Charlottenburg, HRB 209739 B >> > > > > > >> > > > > > *m:* +491603708037 >> > > > > > >> > > > > > *w:* aiven.io *e:* matthew.dedetr...@aiven.io >> > > > > > >> > > > > >> > > > >> > > >> > > >> > > -- >> > > >> > > Matthew de Detrich >> > > >> > > *Aiven Deutschland GmbH* >> > > >> > > Immanuelkirchstraße 26, 10405 Berlin >> > > >> > > Amtsgericht Charlottenburg, HRB 209739 B >> > > >> > > Geschäftsführer: Oskari Saarenmaa & Hannu Valtonen >> > > >> > > *m:* +491603708037 >> > > >> > > *w:* aiven.io *e:* matthew.dedetr...@aiven.io >> > > >> > >> >> >> -- >> >> Matthew de Detrich >> >> *Aiven Deutschland GmbH* >> >> Immanuelkirchstraße 26, 10405 Berlin >> >> Amtsgericht Charlottenburg, HRB 209739 B >> >> Geschäftsführer: Oskari Saarenmaa & Hannu Valtonen >> >> *m:* +491603708037 >> >> *w:* aiven.io *e:* matthew.dedetr...@aiven.io >> > > > -- > > Josep Prat > > *Aiven Deutschland GmbH* > > Immanuelkirchstraße 26, 10405 Berlin > > Amtsgericht Charlottenburg, HRB 209739 B > > Geschäftsführer: Oskari Saarenmaa & Hannu Valtonen > > *m:* +491715557497 > > *w:* aiven.io > > *e:* josep.p...@aiven.io >
