On Sat, Mar 23, 2019, at 18:41, nathank...@gmail.com wrote: > > > On 2019/01/28 02:26:31, n...@afshartous.com wrote: > > Hi Sönke, > > > > Thanks for taking the time to review. I’ve put KIP-349 into hibernation. > > > > Thanks also to everyone who participated in the discussion. > > > > Best regards, > > -- > > Nick > > > > > On Jan 25, 2019, at 5:51 AM, Sönke Liebau > > > <soenke.lie...@opencore.com.INVALID> wrote: > > > > > > a bit late to the party, sorry. I recently spent some time looking > > > into this / a similar issue [1]. > > > After some investigation and playing around with settings I think that > > > the benefit that could be gained from this is somewhat limited and > > > probably outweighed by the implementation effort. > > > > > > The consumer internal are already geared towards treating partitions > > > fairly so that no partition has to wait an undue amount of time and > > > this can be further tuned for latency over throughput. Additionally, > > > if this is a large issue for someone, there is always the option of > > > having a dedicated consumer reading only from the control topic, which > > > would mean that messages from that topic are received "immediately". > > > For a Kafka Streams job it would probably make sense to create two > > > input streams and then merging those as a first step. > > > > > > I think with these knobs a fairly large amount of flexibility can be > > > achieved so that there is no urgent need to implement priorities. > > > > > > So my personal preference would be to set this KIP to dormant for now. > > > > > > > > > > > > > Hello Nick, > > I'm extremely new to Kafka, but I was attempting to set up a per-topic > priority application, and ended up finding this thread. I'm having > difficulty seeing how one can implement it with pause/resume. Would you > elaborate? > > Since those operations are per-partition, and when you stop a > partition, it attempts to re-balance, I would need to stop all > partitions. Even then, it would try to finish the current transactions > instead of immediately putting it on hold and processing other topics.
Hi nathankski, Calling pause() on a partition doesn't trigger a re-balance or try to finish the current transactions. It just means that you won't get more records for that partition until you call resume() on it. > > It also looks like in order to determine if I had received messages > from the pri-1 topic, I would need to loop through all records, and > ignore those that weren't pri-1 until a poll failed to retrieve any, > which seems like it would screw up the other topics. One way to do this would be to have two threads. The first thread calls poll() on the Kafka consumer. It puts the records it retrieves into a PriorityBlockingQueue. Records from pri-1 have the priority within the queue. The second thread retrieves records from the queue. pri-1 records will always be pulled out of the PriorityBlockingQueue ahead of any other records, so they will be processed first. If the priority queue gets too big, you pause partitions until thread 2 can clear the backlog. The low-priority partition is paused first. best, Colin > > Thank you, > > Nathan >
