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 > > > > > > > > > > > > > > > > > > > > > > > > > >
