Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Hi there Kafka developers,

I am currently trying to find a solution to an issue that has been
manifesting itself in the Akka streams implementation of the Kafka
connector. When it comes to consuming messages, the implementation relies
heavily on the fact that we can pause and resume partitions. In some
situations when a single consumer instance is shared among several streams,
we might end up with frequently pausing and unpausing a set of topic
partitions, which is the main facility that allows us to implement back
pressure. This however has certain disadvantages, especially when there are
two consumers that differ in terms of processing speed.

To articulate the issue more clearly, imagine that a consumer maintains
assignments for two topic partitions *TP1* and *TP2*. This consumer is
shared by two streams - S1 and S2. So effectively when we have demand from
only one of the streams - *S1*, we will pause one of the topic partitions
*TP2* and call *poll()* on the consumer to only retrieve the records for
the demanded topic partition - *TP1*. The result of that is all the records
that have been prefetched for *TP2* are now thrown away by the fetcher ("*Not
returning fetched records for assigned partition TP2 since it is no longer
fetchable"*). If we extrapolate that to multiple streams sharing the same
consumer, we might quickly end up in a situation where we throw prefetched
data quite often. This does not seem like the most efficient approach and
in fact produces quite a lot of overlapping fetch requests as illustrated
in the following issue:

https://github.com/akka/alpakka-kafka/issues/549

I am writing this email to get some initial opinion on a KIP I was thinking
about. What if we give the clients of the Consumer API a bit more control
of what to do with this prefetched data. Two options I am wondering about:

1. Introduce a configuration setting, such as*
"return-prefetched-data-for-paused-topic-partitions = false"* (have to
think of a better name), which when set to true will return what is
prefetched instead of throwing it away on calling *poll()*. Since this is
amount of data that is bounded by the maximum size of the prefetch, we can
control what is the most amount of records returned. The client of the
consumer API can then be responsible for keeping that data around and use
it when appropriate (i.e. when demand is present)

2. Introduce a facility to pass in a buffer into which the prefetched
records are drained when poll is called and paused partitions have some
prefetched records.

Any opinions on the matter are welcome. Thanks a lot !

Zahari Dichev

Re: Throwing away prefetched records optimisation.

[Colin McCabe <cm...@apache.org>]

Ryanne Dolan <ry...@gmail.com> wrote:
> It sounds to me like this problem is due to Akka attempting to implement
> additional backpressure on top of the Consumer API. I'd suggest they not do
> that, and then this problem goes away.

Imagine a very simple case where you want to consume from three partitions at about the same rate, but the messages in those partitions have different average sizes.  You can't keep the consumption rate the same without occassionally pausing the fast partitions.

I think we should encourage people to use pause and resume when it's appropriate.  After all, Kafka Streams uses pause and resume, for much the same reasons.  It's demonstrably a simple and useful API.  There's nothing wrong with frameworks implementing backpressure.

Jan Filipiak <Ja...@trivago.com> wrote:
 > The idea for you would be that Messagechooser could hang on to the 
 > prefetched messages.
 > 
 > ccing cmccabe@apache.org
 > 
 > @Collin
 > just for you to see that MessageChooser is a powerfull abstraction.

Samza has a lot of powerful abstractions.  However, in this case, you can get everything you need just by using pause and resume on partitions.  And then Zahari can implement some powerful abstractions of his own, in Akka Streams. :)

Matthias J. Sax <ma...@confluent.io> wrote:
> Heartbeats are send by a backgroud thread since 0.10.0.0. And there is a
> second configuration `max.poll.interval.ms` that you can increase for a
> slow consumer. Kafka Streams sets it to `Integer.MAX_VALUE` for example.

Right, that's a very important point.  Heartbeats are sent by a background thread now.

Zahari Dichev <za...@gmail.com> wrote:
> Just a mall thing to mention though, blocking on
> the poll forever is not something that I am convinced we would like to do.

There is a KafkaConsumer#wakeup method which might help here.
>     * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.

best,
Colin


On Sun, Oct 21, 2018, at 14:59, Matthias J. Sax wrote:
> It's spread out multiple classes...
> 
> Good starting point is here:
> https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L806
> 
> It implements the main-loop that polls, addRecordsToTasks() (ie, put the
> into buffers), and processes record.
> 
> pause/resume is done here:
> 
> https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L720
> 
> and
> 
> https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L362
> 
> 
> Not blocking forever makes sense of course. I did not mean in literally.
> However, longer poll() blocking times, should allow you to drain the
> consumer buffer better.
> 
> 
> -Matthias
> 
> On 10/21/18 12:54 PM, Zahari Dichev wrote:
> > Thanks for your feedback Matthias, Do you think you can point me to the
> > part where Kafka streams deals with all of that so I can take a look. Will
> > try and see whether your suggested approach works for us before trying to
> > argue my point further. Just a mall thing to mention though, blocking on
> > the poll forever is not something that I am convinced we would like to do.
> > 
> > Zahari
> > 
> > On Sun, Oct 21, 2018 at 10:11 PM Matthias J. Sax <ma...@confluent.io>
> > wrote:
> > 
> >>> You have one consumer that is quite
> >>>> slow so lets say it call poll every 5 seconds, while you need to call
> >> poll
> >>>> every 1 second to issue a heartbeat (these are made up numbers of
> >> course).
> >>
> >> Heartbeats are send by a backgroud thread since 0.10.0.0. And there is a
> >> second configuration `max.poll.interval.ms` that you can increase for a
> >> slow consumer. Kafka Streams sets it to `Integer.MAX_VALUE` for example.
> >>
> >> For the remainder, I agree, that using a buffer on top might not a
> >> perfect solution. That is, what I meant by:
> >>
> >>> (not 100% percent guarantee,
> >>> depending on fetch size and max.poll.record etc, but with high
> >>> probability)
> >>
> >> However, I believe, that if you configure the consumer accordingly, you
> >> can to drain the fetch buffer if you block on `poll()` forever.
> >>
> >> I have to admit, than I am not familiar with the details of pipelining
> >> fetch requests thought. The general idea is, to make sure to drain the
> >> internal buffer of the consumer, before you call `pause()`.
> >>
> >> Curious to to learn why this would not work? How does the pipelining of
> >> fetch requests works in detail?
> >>
> >>
> >> -Matthias
> >>
> >>
> >> On 10/19/18 1:36 PM, Zahari Dichev wrote:
> >>> Hi there Matthias,
> >>>
> >>> Very useful thoughts indeed. I have considered the exact same approach
> >> but
> >>> what worries me a bit is that I do not think that will certainly solve
> >> the
> >>> issue. Imagine the following situation. You have one consumer that is
> >> quite
> >>> slow so lets say it call poll every 5 seconds, while you need to call
> >> poll
> >>> every 1 second to issue a heartbeat (these are made up numbers of
> >> course).
> >>>
> >>> So our consumer calls poll at T0 grabs some data and puts it in a buffer
> >>> and calls pause on the topic partition. The fetcher tries to pipeline and
> >>> issues a fetch request and at some point the data arrives. At that point
> >> we
> >>> have some data in the buffer and we can do whatever we want with it, but
> >>> there is also some data living within the consumer/fetcher. Approximately
> >>> one second later we call poll again because we need to. We are not
> >> getting
> >>> any data because our partition is paused and this is good because if we
> >> got
> >>> data we would not know what to do with it as our client is still busy
> >>> crunching the data from the first poll. So far so good. What happens
> >> though
> >>> is that the pre-fetched data gets thrown away upon calling poll as its
> >> "no
> >>> longer fetchable...".  Maybe I am not fully understanding your suggested
> >>> approach, but I dont think it would solve this problem.
> >>>
> >>> Zahari
> >>>
> >>> On Fri, Oct 19, 2018 at 8:10 PM Matthias J. Sax <ma...@confluent.io>
> >>> wrote:
> >>>
> >>>> Just my 2 cents.
> >>>>
> >>>> I am not 100% sure if we would need to change the consumer for this.
> >>>> While I still think, that KIP-349 might be valuable, it seems to be
> >>>> complementary/orthogonal to the issue discussed here.
> >>>>
> >>>> For Kafka Streams, we have a related scenario and what Kafka Streams
> >>>> does is, to add its own buffer on top of the consumer. Thus, for each
> >>>> `poll()` all data is put into this buffer, and now Streams can decide
> >>>> which record to process first. For buffers that have data, we can call
> >>>> `pause()` without loosing fetched data (not 100% percent guarantee,
> >>>> depending on fetch size and max.poll.record etc, but with high
> >>>> probability) and if a buffer gets empty we `resume()` partitions.
> >>>>
> >>>> As Akka Streams builds on top of the consumer it could implement a
> >>>> similar pattern. Of course, one cannot use `auto.commit` on the
> >>>> consumer, but commits need to be manged manually, (ie, only data that
> >>>> was take out of the buffer and actually was processed can be committed).
> >>>>
> >>>> For the `MessageChooser` idea, I also still think it might be useful,
> >>>> but it's unclear to me if this should be a consumer feature or build on
> >>>> top of the consumer (maybe it could be a Streams feature, as Streams is
> >>>> build on top of the consumer). Thoughts?
> >>>>
> >>>>
> >>>> -Matthias
> >>>>
> >>>> On 10/18/18 9:27 AM, Jan Filipiak wrote:
> >>>>> The idea for you would be that Messagechooser could hang on to the
> >>>>> prefetched messages.
> >>>>>
> >>>>> ccing cmccabe@apache.org
> >>>>>
> >>>>> @Collin
> >>>>> just for you to see that MessageChooser is a powerfull abstraction.
> >>>>>
> >>>>> :)
> >>>>>
> >>>>> Best jan
> >>>>>
> >>>>> On 18.10.2018 13:59, Zahari Dichev wrote:
> >>>>>> Jan,
> >>>>>>
> >>>>>> Quite insightful indeed. I think your propositions are valid.
> >>>>>>
> >>>>>> Ryanne,
> >>>>>>
> >>>>>> I understand that consumers are using a pull model... And yes, indeed
> >>>> if a
> >>>>>> consumer is not ready for more records it surely should not call poll.
> >>>>>> Except that it needs to do so periodically in order to indicate that
> >> its
> >>>>>> live. Forget about the "backpressure", I guess I was wrong with
> >> phrasing
> >>>>>> this so lets not get caught up on it.
> >>>>>>
> >>>>>> You say pause/resume can be used to prioritise certain
> >> topics/partitions
> >>>>>> over others. And indeed this is the case. So instead of thinking about
> >>>> it
> >>>>>> in terms of backpressure, lets put it in a different way. The Akka
> >>>> streams
> >>>>>> connector would like to prioritise certain topics over others, using
> >>>> once
> >>>>>> consumer instance. On top of that, add the detail that the priorities
> >>>>>> change quite frequently (which translates to calling pause/resume
> >>>>>> frequently). So all that being said, what would be a proper way to
> >>>> handle
> >>>>>> the situation without throwing the pre-fetched records away when
> >> calling
> >>>>>> poll on a consumer that happens to have a topic that was recently
> >> paused
> >>>>>> (and that might be un-paused soon )? Am I the only one who considers
> >>>> that
> >>>>>> an actual problem with the use os pause/resume ? Not sure how to
> >> explain
> >>>>>> the situation in a better way..
> >>>>>>
> >>>>>> Zahari
> >>>>>>
> >>>>>>
> >>>>>> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <zaharidichev@gmail.com
> >>>
> >>>>>> wrote:
> >>>>>>
> >>>>>>> Thanks a lot Jan,
> >>>>>>>
> >>>>>>> I will read it.
> >>>>>>>
> >>>>>>> Zahari
> >>>>>>>
> >>>>>>> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <
> >> Jan.Filipiak@trivago.com
> >>>>>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>>> especially my suggestions ;)
> >>>>>>>>
> >>>>>>>> On 18.10.2018 08:30, Jan Filipiak wrote:
> >>>>>>>>> Hi Zahari,
> >>>>>>>>>
> >>>>>>>>> would you be willing to scan through the KIP-349 discussion a
> >> little?
> >>>>>>>>> I think it has suggestions that could be interesting for you
> >>>>>>>>>
> >>>>>>>>> Best Jan
> >>>>>>>>>
> >>>>>>>>> On 16.10.2018 09:29, Zahari Dichev wrote:
> >>>>>>>>>> Hi there Kafka developers,
> >>>>>>>>>>
> >>>>>>>>>> I am currently trying to find a solution to an issue that has been
> >>>>>>>>>> manifesting itself in the Akka streams implementation of the Kafka
> >>>>>>>>>> connector. When it comes to consuming messages, the implementation
> >>>>>>>> relies
> >>>>>>>>>> heavily on the fact that we can pause and resume partitions. In
> >> some
> >>>>>>>>>> situations when a single consumer instance is shared among several
> >>>>>>>>>> streams,
> >>>>>>>>>> we might end up with frequently pausing and unpausing a set of
> >> topic
> >>>>>>>>>> partitions, which is the main facility that allows us to implement
> >>>> back
> >>>>>>>>>> pressure. This however has certain disadvantages, especially when
> >>>>>>>>>> there are
> >>>>>>>>>> two consumers that differ in terms of processing speed.
> >>>>>>>>>>
> >>>>>>>>>> To articulate the issue more clearly, imagine that a consumer
> >>>> maintains
> >>>>>>>>>> assignments for two topic partitions *TP1* and *TP2*. This
> >> consumer
> >>>> is
> >>>>>>>>>> shared by two streams - S1 and S2. So effectively when we have
> >>>> demand
> >>>>>>>>>> from
> >>>>>>>>>> only one of the streams - *S1*, we will pause one of the topic
> >>>>>>>> partitions
> >>>>>>>>>> *TP2* and call *poll()* on the consumer to only retrieve the
> >> records
> >>>>>>>> for
> >>>>>>>>>> the demanded topic partition - *TP1*. The result of that is all
> >> the
> >>>>>>>>>> records
> >>>>>>>>>> that have been prefetched for *TP2* are now thrown away by the
> >>>> fetcher
> >>>>>>>>>> ("*Not
> >>>>>>>>>> returning fetched records for assigned partition TP2 since it is
> >> no
> >>>>>>>>>> longer
> >>>>>>>>>> fetchable"*). If we extrapolate that to multiple streams sharing
> >> the
> >>>>>>>> same
> >>>>>>>>>> consumer, we might quickly end up in a situation where we throw
> >>>>>>>>>> prefetched
> >>>>>>>>>> data quite often. This does not seem like the most efficient
> >>>> approach
> >>>>>>>> and
> >>>>>>>>>> in fact produces quite a lot of overlapping fetch requests as
> >>>>>>>> illustrated
> >>>>>>>>>> in the following issue:
> >>>>>>>>>>
> >>>>>>>>>> https://github.com/akka/alpakka-kafka/issues/549
> >>>>>>>>>>
> >>>>>>>>>> I am writing this email to get some initial opinion on a KIP I was
> >>>>>>>>>> thinking
> >>>>>>>>>> about. What if we give the clients of the Consumer API a bit more
> >>>>>>>> control
> >>>>>>>>>> of what to do with this prefetched data. Two options I am
> >> wondering
> >>>>>>>>>> about:
> >>>>>>>>>>
> >>>>>>>>>> 1. Introduce a configuration setting, such as*
> >>>>>>>>>> "return-prefetched-data-for-paused-topic-partitions = false"*
> >> (have
> >>>> to
> >>>>>>>>>> think of a better name), which when set to true will return what
> >> is
> >>>>>>>>>> prefetched instead of throwing it away on calling *poll()*. Since
> >>>> this
> >>>>>>>> is
> >>>>>>>>>> amount of data that is bounded by the maximum size of the
> >> prefetch,
> >>>> we
> >>>>>>>>>> can
> >>>>>>>>>> control what is the most amount of records returned. The client of
> >>>> the
> >>>>>>>>>> consumer API can then be responsible for keeping that data around
> >>>> and
> >>>>>>>> use
> >>>>>>>>>> it when appropriate (i.e. when demand is present)
> >>>>>>>>>>
> >>>>>>>>>> 2. Introduce a facility to pass in a buffer into which the
> >>>> prefetched
> >>>>>>>>>> records are drained when poll is called and paused partitions have
> >>>> some
> >>>>>>>>>> prefetched records.
> >>>>>>>>>>
> >>>>>>>>>> Any opinions on the matter are welcome. Thanks a lot !
> >>>>>>>>>>
> >>>>>>>>>> Zahari Dichev
> >>>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>>
> >>>>
> >>>
> >>
> >>
> > 
> 
> Email had 1 attachment:
> + signature.asc
>   1k (application/pgp-signature)

Re: Throwing away prefetched records optimisation.

["Matthias J. Sax" <ma...@confluent.io>]

It's spread out multiple classes...

Good starting point is here:
https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L806

It implements the main-loop that polls, addRecordsToTasks() (ie, put the
into buffers), and processes record.

pause/resume is done here:

https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L720

and

https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java#L362


Not blocking forever makes sense of course. I did not mean in literally.
However, longer poll() blocking times, should allow you to drain the
consumer buffer better.


-Matthias

On 10/21/18 12:54 PM, Zahari Dichev wrote:
> Thanks for your feedback Matthias, Do you think you can point me to the
> part where Kafka streams deals with all of that so I can take a look. Will
> try and see whether your suggested approach works for us before trying to
> argue my point further. Just a mall thing to mention though, blocking on
> the poll forever is not something that I am convinced we would like to do.
> 
> Zahari
> 
> On Sun, Oct 21, 2018 at 10:11 PM Matthias J. Sax <ma...@confluent.io>
> wrote:
> 
>>> You have one consumer that is quite
>>>> slow so lets say it call poll every 5 seconds, while you need to call
>> poll
>>>> every 1 second to issue a heartbeat (these are made up numbers of
>> course).
>>
>> Heartbeats are send by a backgroud thread since 0.10.0.0. And there is a
>> second configuration `max.poll.interval.ms` that you can increase for a
>> slow consumer. Kafka Streams sets it to `Integer.MAX_VALUE` for example.
>>
>> For the remainder, I agree, that using a buffer on top might not a
>> perfect solution. That is, what I meant by:
>>
>>> (not 100% percent guarantee,
>>> depending on fetch size and max.poll.record etc, but with high
>>> probability)
>>
>> However, I believe, that if you configure the consumer accordingly, you
>> can to drain the fetch buffer if you block on `poll()` forever.
>>
>> I have to admit, than I am not familiar with the details of pipelining
>> fetch requests thought. The general idea is, to make sure to drain the
>> internal buffer of the consumer, before you call `pause()`.
>>
>> Curious to to learn why this would not work? How does the pipelining of
>> fetch requests works in detail?
>>
>>
>> -Matthias
>>
>>
>> On 10/19/18 1:36 PM, Zahari Dichev wrote:
>>> Hi there Matthias,
>>>
>>> Very useful thoughts indeed. I have considered the exact same approach
>> but
>>> what worries me a bit is that I do not think that will certainly solve
>> the
>>> issue. Imagine the following situation. You have one consumer that is
>> quite
>>> slow so lets say it call poll every 5 seconds, while you need to call
>> poll
>>> every 1 second to issue a heartbeat (these are made up numbers of
>> course).
>>>
>>> So our consumer calls poll at T0 grabs some data and puts it in a buffer
>>> and calls pause on the topic partition. The fetcher tries to pipeline and
>>> issues a fetch request and at some point the data arrives. At that point
>> we
>>> have some data in the buffer and we can do whatever we want with it, but
>>> there is also some data living within the consumer/fetcher. Approximately
>>> one second later we call poll again because we need to. We are not
>> getting
>>> any data because our partition is paused and this is good because if we
>> got
>>> data we would not know what to do with it as our client is still busy
>>> crunching the data from the first poll. So far so good. What happens
>> though
>>> is that the pre-fetched data gets thrown away upon calling poll as its
>> "no
>>> longer fetchable...".  Maybe I am not fully understanding your suggested
>>> approach, but I dont think it would solve this problem.
>>>
>>> Zahari
>>>
>>> On Fri, Oct 19, 2018 at 8:10 PM Matthias J. Sax <ma...@confluent.io>
>>> wrote:
>>>
>>>> Just my 2 cents.
>>>>
>>>> I am not 100% sure if we would need to change the consumer for this.
>>>> While I still think, that KIP-349 might be valuable, it seems to be
>>>> complementary/orthogonal to the issue discussed here.
>>>>
>>>> For Kafka Streams, we have a related scenario and what Kafka Streams
>>>> does is, to add its own buffer on top of the consumer. Thus, for each
>>>> `poll()` all data is put into this buffer, and now Streams can decide
>>>> which record to process first. For buffers that have data, we can call
>>>> `pause()` without loosing fetched data (not 100% percent guarantee,
>>>> depending on fetch size and max.poll.record etc, but with high
>>>> probability) and if a buffer gets empty we `resume()` partitions.
>>>>
>>>> As Akka Streams builds on top of the consumer it could implement a
>>>> similar pattern. Of course, one cannot use `auto.commit` on the
>>>> consumer, but commits need to be manged manually, (ie, only data that
>>>> was take out of the buffer and actually was processed can be committed).
>>>>
>>>> For the `MessageChooser` idea, I also still think it might be useful,
>>>> but it's unclear to me if this should be a consumer feature or build on
>>>> top of the consumer (maybe it could be a Streams feature, as Streams is
>>>> build on top of the consumer). Thoughts?
>>>>
>>>>
>>>> -Matthias
>>>>
>>>> On 10/18/18 9:27 AM, Jan Filipiak wrote:
>>>>> The idea for you would be that Messagechooser could hang on to the
>>>>> prefetched messages.
>>>>>
>>>>> ccing cmccabe@apache.org
>>>>>
>>>>> @Collin
>>>>> just for you to see that MessageChooser is a powerfull abstraction.
>>>>>
>>>>> :)
>>>>>
>>>>> Best jan
>>>>>
>>>>> On 18.10.2018 13:59, Zahari Dichev wrote:
>>>>>> Jan,
>>>>>>
>>>>>> Quite insightful indeed. I think your propositions are valid.
>>>>>>
>>>>>> Ryanne,
>>>>>>
>>>>>> I understand that consumers are using a pull model... And yes, indeed
>>>> if a
>>>>>> consumer is not ready for more records it surely should not call poll.
>>>>>> Except that it needs to do so periodically in order to indicate that
>> its
>>>>>> live. Forget about the "backpressure", I guess I was wrong with
>> phrasing
>>>>>> this so lets not get caught up on it.
>>>>>>
>>>>>> You say pause/resume can be used to prioritise certain
>> topics/partitions
>>>>>> over others. And indeed this is the case. So instead of thinking about
>>>> it
>>>>>> in terms of backpressure, lets put it in a different way. The Akka
>>>> streams
>>>>>> connector would like to prioritise certain topics over others, using
>>>> once
>>>>>> consumer instance. On top of that, add the detail that the priorities
>>>>>> change quite frequently (which translates to calling pause/resume
>>>>>> frequently). So all that being said, what would be a proper way to
>>>> handle
>>>>>> the situation without throwing the pre-fetched records away when
>> calling
>>>>>> poll on a consumer that happens to have a topic that was recently
>> paused
>>>>>> (and that might be un-paused soon )? Am I the only one who considers
>>>> that
>>>>>> an actual problem with the use os pause/resume ? Not sure how to
>> explain
>>>>>> the situation in a better way..
>>>>>>
>>>>>> Zahari
>>>>>>
>>>>>>
>>>>>> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <zaharidichev@gmail.com
>>>
>>>>>> wrote:
>>>>>>
>>>>>>> Thanks a lot Jan,
>>>>>>>
>>>>>>> I will read it.
>>>>>>>
>>>>>>> Zahari
>>>>>>>
>>>>>>> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <
>> Jan.Filipiak@trivago.com
>>>>>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> especially my suggestions ;)
>>>>>>>>
>>>>>>>> On 18.10.2018 08:30, Jan Filipiak wrote:
>>>>>>>>> Hi Zahari,
>>>>>>>>>
>>>>>>>>> would you be willing to scan through the KIP-349 discussion a
>> little?
>>>>>>>>> I think it has suggestions that could be interesting for you
>>>>>>>>>
>>>>>>>>> Best Jan
>>>>>>>>>
>>>>>>>>> On 16.10.2018 09:29, Zahari Dichev wrote:
>>>>>>>>>> Hi there Kafka developers,
>>>>>>>>>>
>>>>>>>>>> I am currently trying to find a solution to an issue that has been
>>>>>>>>>> manifesting itself in the Akka streams implementation of the Kafka
>>>>>>>>>> connector. When it comes to consuming messages, the implementation
>>>>>>>> relies
>>>>>>>>>> heavily on the fact that we can pause and resume partitions. In
>> some
>>>>>>>>>> situations when a single consumer instance is shared among several
>>>>>>>>>> streams,
>>>>>>>>>> we might end up with frequently pausing and unpausing a set of
>> topic
>>>>>>>>>> partitions, which is the main facility that allows us to implement
>>>> back
>>>>>>>>>> pressure. This however has certain disadvantages, especially when
>>>>>>>>>> there are
>>>>>>>>>> two consumers that differ in terms of processing speed.
>>>>>>>>>>
>>>>>>>>>> To articulate the issue more clearly, imagine that a consumer
>>>> maintains
>>>>>>>>>> assignments for two topic partitions *TP1* and *TP2*. This
>> consumer
>>>> is
>>>>>>>>>> shared by two streams - S1 and S2. So effectively when we have
>>>> demand
>>>>>>>>>> from
>>>>>>>>>> only one of the streams - *S1*, we will pause one of the topic
>>>>>>>> partitions
>>>>>>>>>> *TP2* and call *poll()* on the consumer to only retrieve the
>> records
>>>>>>>> for
>>>>>>>>>> the demanded topic partition - *TP1*. The result of that is all
>> the
>>>>>>>>>> records
>>>>>>>>>> that have been prefetched for *TP2* are now thrown away by the
>>>> fetcher
>>>>>>>>>> ("*Not
>>>>>>>>>> returning fetched records for assigned partition TP2 since it is
>> no
>>>>>>>>>> longer
>>>>>>>>>> fetchable"*). If we extrapolate that to multiple streams sharing
>> the
>>>>>>>> same
>>>>>>>>>> consumer, we might quickly end up in a situation where we throw
>>>>>>>>>> prefetched
>>>>>>>>>> data quite often. This does not seem like the most efficient
>>>> approach
>>>>>>>> and
>>>>>>>>>> in fact produces quite a lot of overlapping fetch requests as
>>>>>>>> illustrated
>>>>>>>>>> in the following issue:
>>>>>>>>>>
>>>>>>>>>> https://github.com/akka/alpakka-kafka/issues/549
>>>>>>>>>>
>>>>>>>>>> I am writing this email to get some initial opinion on a KIP I was
>>>>>>>>>> thinking
>>>>>>>>>> about. What if we give the clients of the Consumer API a bit more
>>>>>>>> control
>>>>>>>>>> of what to do with this prefetched data. Two options I am
>> wondering
>>>>>>>>>> about:
>>>>>>>>>>
>>>>>>>>>> 1. Introduce a configuration setting, such as*
>>>>>>>>>> "return-prefetched-data-for-paused-topic-partitions = false"*
>> (have
>>>> to
>>>>>>>>>> think of a better name), which when set to true will return what
>> is
>>>>>>>>>> prefetched instead of throwing it away on calling *poll()*. Since
>>>> this
>>>>>>>> is
>>>>>>>>>> amount of data that is bounded by the maximum size of the
>> prefetch,
>>>> we
>>>>>>>>>> can
>>>>>>>>>> control what is the most amount of records returned. The client of
>>>> the
>>>>>>>>>> consumer API can then be responsible for keeping that data around
>>>> and
>>>>>>>> use
>>>>>>>>>> it when appropriate (i.e. when demand is present)
>>>>>>>>>>
>>>>>>>>>> 2. Introduce a facility to pass in a buffer into which the
>>>> prefetched
>>>>>>>>>> records are drained when poll is called and paused partitions have
>>>> some
>>>>>>>>>> prefetched records.
>>>>>>>>>>
>>>>>>>>>> Any opinions on the matter are welcome. Thanks a lot !
>>>>>>>>>>
>>>>>>>>>> Zahari Dichev
>>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>
>>>>
>>>
>>
>>
> 

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Thanks for your feedback Matthias, Do you think you can point me to the
part where Kafka streams deals with all of that so I can take a look. Will
try and see whether your suggested approach works for us before trying to
argue my point further. Just a mall thing to mention though, blocking on
the poll forever is not something that I am convinced we would like to do.

Zahari

On Sun, Oct 21, 2018 at 10:11 PM Matthias J. Sax <ma...@confluent.io>
wrote:

> > You have one consumer that is quite
> >> slow so lets say it call poll every 5 seconds, while you need to call
> poll
> >> every 1 second to issue a heartbeat (these are made up numbers of
> course).
>
> Heartbeats are send by a backgroud thread since 0.10.0.0. And there is a
> second configuration `max.poll.interval.ms` that you can increase for a
> slow consumer. Kafka Streams sets it to `Integer.MAX_VALUE` for example.
>
> For the remainder, I agree, that using a buffer on top might not a
> perfect solution. That is, what I meant by:
>
> > (not 100% percent guarantee,
> > depending on fetch size and max.poll.record etc, but with high
> > probability)
>
> However, I believe, that if you configure the consumer accordingly, you
> can to drain the fetch buffer if you block on `poll()` forever.
>
> I have to admit, than I am not familiar with the details of pipelining
> fetch requests thought. The general idea is, to make sure to drain the
> internal buffer of the consumer, before you call `pause()`.
>
> Curious to to learn why this would not work? How does the pipelining of
> fetch requests works in detail?
>
>
> -Matthias
>
>
> On 10/19/18 1:36 PM, Zahari Dichev wrote:
> > Hi there Matthias,
> >
> > Very useful thoughts indeed. I have considered the exact same approach
> but
> > what worries me a bit is that I do not think that will certainly solve
> the
> > issue. Imagine the following situation. You have one consumer that is
> quite
> > slow so lets say it call poll every 5 seconds, while you need to call
> poll
> > every 1 second to issue a heartbeat (these are made up numbers of
> course).
> >
> > So our consumer calls poll at T0 grabs some data and puts it in a buffer
> > and calls pause on the topic partition. The fetcher tries to pipeline and
> > issues a fetch request and at some point the data arrives. At that point
> we
> > have some data in the buffer and we can do whatever we want with it, but
> > there is also some data living within the consumer/fetcher. Approximately
> > one second later we call poll again because we need to. We are not
> getting
> > any data because our partition is paused and this is good because if we
> got
> > data we would not know what to do with it as our client is still busy
> > crunching the data from the first poll. So far so good. What happens
> though
> > is that the pre-fetched data gets thrown away upon calling poll as its
> "no
> > longer fetchable...".  Maybe I am not fully understanding your suggested
> > approach, but I dont think it would solve this problem.
> >
> > Zahari
> >
> > On Fri, Oct 19, 2018 at 8:10 PM Matthias J. Sax <ma...@confluent.io>
> > wrote:
> >
> >> Just my 2 cents.
> >>
> >> I am not 100% sure if we would need to change the consumer for this.
> >> While I still think, that KIP-349 might be valuable, it seems to be
> >> complementary/orthogonal to the issue discussed here.
> >>
> >> For Kafka Streams, we have a related scenario and what Kafka Streams
> >> does is, to add its own buffer on top of the consumer. Thus, for each
> >> `poll()` all data is put into this buffer, and now Streams can decide
> >> which record to process first. For buffers that have data, we can call
> >> `pause()` without loosing fetched data (not 100% percent guarantee,
> >> depending on fetch size and max.poll.record etc, but with high
> >> probability) and if a buffer gets empty we `resume()` partitions.
> >>
> >> As Akka Streams builds on top of the consumer it could implement a
> >> similar pattern. Of course, one cannot use `auto.commit` on the
> >> consumer, but commits need to be manged manually, (ie, only data that
> >> was take out of the buffer and actually was processed can be committed).
> >>
> >> For the `MessageChooser` idea, I also still think it might be useful,
> >> but it's unclear to me if this should be a consumer feature or build on
> >> top of the consumer (maybe it could be a Streams feature, as Streams is
> >> build on top of the consumer). Thoughts?
> >>
> >>
> >> -Matthias
> >>
> >> On 10/18/18 9:27 AM, Jan Filipiak wrote:
> >>> The idea for you would be that Messagechooser could hang on to the
> >>> prefetched messages.
> >>>
> >>> ccing cmccabe@apache.org
> >>>
> >>> @Collin
> >>> just for you to see that MessageChooser is a powerfull abstraction.
> >>>
> >>> :)
> >>>
> >>> Best jan
> >>>
> >>> On 18.10.2018 13:59, Zahari Dichev wrote:
> >>>> Jan,
> >>>>
> >>>> Quite insightful indeed. I think your propositions are valid.
> >>>>
> >>>> Ryanne,
> >>>>
> >>>> I understand that consumers are using a pull model... And yes, indeed
> >> if a
> >>>> consumer is not ready for more records it surely should not call poll.
> >>>> Except that it needs to do so periodically in order to indicate that
> its
> >>>> live. Forget about the "backpressure", I guess I was wrong with
> phrasing
> >>>> this so lets not get caught up on it.
> >>>>
> >>>> You say pause/resume can be used to prioritise certain
> topics/partitions
> >>>> over others. And indeed this is the case. So instead of thinking about
> >> it
> >>>> in terms of backpressure, lets put it in a different way. The Akka
> >> streams
> >>>> connector would like to prioritise certain topics over others, using
> >> once
> >>>> consumer instance. On top of that, add the detail that the priorities
> >>>> change quite frequently (which translates to calling pause/resume
> >>>> frequently). So all that being said, what would be a proper way to
> >> handle
> >>>> the situation without throwing the pre-fetched records away when
> calling
> >>>> poll on a consumer that happens to have a topic that was recently
> paused
> >>>> (and that might be un-paused soon )? Am I the only one who considers
> >> that
> >>>> an actual problem with the use os pause/resume ? Not sure how to
> explain
> >>>> the situation in a better way..
> >>>>
> >>>> Zahari
> >>>>
> >>>>
> >>>> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <zaharidichev@gmail.com
> >
> >>>> wrote:
> >>>>
> >>>>> Thanks a lot Jan,
> >>>>>
> >>>>> I will read it.
> >>>>>
> >>>>> Zahari
> >>>>>
> >>>>> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <
> Jan.Filipiak@trivago.com
> >>>
> >>>>> wrote:
> >>>>>
> >>>>>> especially my suggestions ;)
> >>>>>>
> >>>>>> On 18.10.2018 08:30, Jan Filipiak wrote:
> >>>>>>> Hi Zahari,
> >>>>>>>
> >>>>>>> would you be willing to scan through the KIP-349 discussion a
> little?
> >>>>>>> I think it has suggestions that could be interesting for you
> >>>>>>>
> >>>>>>> Best Jan
> >>>>>>>
> >>>>>>> On 16.10.2018 09:29, Zahari Dichev wrote:
> >>>>>>>> Hi there Kafka developers,
> >>>>>>>>
> >>>>>>>> I am currently trying to find a solution to an issue that has been
> >>>>>>>> manifesting itself in the Akka streams implementation of the Kafka
> >>>>>>>> connector. When it comes to consuming messages, the implementation
> >>>>>> relies
> >>>>>>>> heavily on the fact that we can pause and resume partitions. In
> some
> >>>>>>>> situations when a single consumer instance is shared among several
> >>>>>>>> streams,
> >>>>>>>> we might end up with frequently pausing and unpausing a set of
> topic
> >>>>>>>> partitions, which is the main facility that allows us to implement
> >> back
> >>>>>>>> pressure. This however has certain disadvantages, especially when
> >>>>>>>> there are
> >>>>>>>> two consumers that differ in terms of processing speed.
> >>>>>>>>
> >>>>>>>> To articulate the issue more clearly, imagine that a consumer
> >> maintains
> >>>>>>>> assignments for two topic partitions *TP1* and *TP2*. This
> consumer
> >> is
> >>>>>>>> shared by two streams - S1 and S2. So effectively when we have
> >> demand
> >>>>>>>> from
> >>>>>>>> only one of the streams - *S1*, we will pause one of the topic
> >>>>>> partitions
> >>>>>>>> *TP2* and call *poll()* on the consumer to only retrieve the
> records
> >>>>>> for
> >>>>>>>> the demanded topic partition - *TP1*. The result of that is all
> the
> >>>>>>>> records
> >>>>>>>> that have been prefetched for *TP2* are now thrown away by the
> >> fetcher
> >>>>>>>> ("*Not
> >>>>>>>> returning fetched records for assigned partition TP2 since it is
> no
> >>>>>>>> longer
> >>>>>>>> fetchable"*). If we extrapolate that to multiple streams sharing
> the
> >>>>>> same
> >>>>>>>> consumer, we might quickly end up in a situation where we throw
> >>>>>>>> prefetched
> >>>>>>>> data quite often. This does not seem like the most efficient
> >> approach
> >>>>>> and
> >>>>>>>> in fact produces quite a lot of overlapping fetch requests as
> >>>>>> illustrated
> >>>>>>>> in the following issue:
> >>>>>>>>
> >>>>>>>> https://github.com/akka/alpakka-kafka/issues/549
> >>>>>>>>
> >>>>>>>> I am writing this email to get some initial opinion on a KIP I was
> >>>>>>>> thinking
> >>>>>>>> about. What if we give the clients of the Consumer API a bit more
> >>>>>> control
> >>>>>>>> of what to do with this prefetched data. Two options I am
> wondering
> >>>>>>>> about:
> >>>>>>>>
> >>>>>>>> 1. Introduce a configuration setting, such as*
> >>>>>>>> "return-prefetched-data-for-paused-topic-partitions = false"*
> (have
> >> to
> >>>>>>>> think of a better name), which when set to true will return what
> is
> >>>>>>>> prefetched instead of throwing it away on calling *poll()*. Since
> >> this
> >>>>>> is
> >>>>>>>> amount of data that is bounded by the maximum size of the
> prefetch,
> >> we
> >>>>>>>> can
> >>>>>>>> control what is the most amount of records returned. The client of
> >> the
> >>>>>>>> consumer API can then be responsible for keeping that data around
> >> and
> >>>>>> use
> >>>>>>>> it when appropriate (i.e. when demand is present)
> >>>>>>>>
> >>>>>>>> 2. Introduce a facility to pass in a buffer into which the
> >> prefetched
> >>>>>>>> records are drained when poll is called and paused partitions have
> >> some
> >>>>>>>> prefetched records.
> >>>>>>>>
> >>>>>>>> Any opinions on the matter are welcome. Thanks a lot !
> >>>>>>>>
> >>>>>>>> Zahari Dichev
> >>>>>>>>
> >>>>>>
> >>>>>
> >>>>
> >>
> >>
> >
>
>

Re: Throwing away prefetched records optimisation.

["Matthias J. Sax" <ma...@confluent.io>]

> You have one consumer that is quite
>> slow so lets say it call poll every 5 seconds, while you need to call poll
>> every 1 second to issue a heartbeat (these are made up numbers of course).

Heartbeats are send by a backgroud thread since 0.10.0.0. And there is a
second configuration `max.poll.interval.ms` that you can increase for a
slow consumer. Kafka Streams sets it to `Integer.MAX_VALUE` for example.

For the remainder, I agree, that using a buffer on top might not a
perfect solution. That is, what I meant by:

> (not 100% percent guarantee,
> depending on fetch size and max.poll.record etc, but with high
> probability)

However, I believe, that if you configure the consumer accordingly, you
can to drain the fetch buffer if you block on `poll()` forever.

I have to admit, than I am not familiar with the details of pipelining
fetch requests thought. The general idea is, to make sure to drain the
internal buffer of the consumer, before you call `pause()`.

Curious to to learn why this would not work? How does the pipelining of
fetch requests works in detail?


-Matthias


On 10/19/18 1:36 PM, Zahari Dichev wrote:
> Hi there Matthias,
> 
> Very useful thoughts indeed. I have considered the exact same approach but
> what worries me a bit is that I do not think that will certainly solve the
> issue. Imagine the following situation. You have one consumer that is quite
> slow so lets say it call poll every 5 seconds, while you need to call poll
> every 1 second to issue a heartbeat (these are made up numbers of course).
> 
> So our consumer calls poll at T0 grabs some data and puts it in a buffer
> and calls pause on the topic partition. The fetcher tries to pipeline and
> issues a fetch request and at some point the data arrives. At that point we
> have some data in the buffer and we can do whatever we want with it, but
> there is also some data living within the consumer/fetcher. Approximately
> one second later we call poll again because we need to. We are not getting
> any data because our partition is paused and this is good because if we got
> data we would not know what to do with it as our client is still busy
> crunching the data from the first poll. So far so good. What happens though
> is that the pre-fetched data gets thrown away upon calling poll as its "no
> longer fetchable...".  Maybe I am not fully understanding your suggested
> approach, but I dont think it would solve this problem.
> 
> Zahari
> 
> On Fri, Oct 19, 2018 at 8:10 PM Matthias J. Sax <ma...@confluent.io>
> wrote:
> 
>> Just my 2 cents.
>>
>> I am not 100% sure if we would need to change the consumer for this.
>> While I still think, that KIP-349 might be valuable, it seems to be
>> complementary/orthogonal to the issue discussed here.
>>
>> For Kafka Streams, we have a related scenario and what Kafka Streams
>> does is, to add its own buffer on top of the consumer. Thus, for each
>> `poll()` all data is put into this buffer, and now Streams can decide
>> which record to process first. For buffers that have data, we can call
>> `pause()` without loosing fetched data (not 100% percent guarantee,
>> depending on fetch size and max.poll.record etc, but with high
>> probability) and if a buffer gets empty we `resume()` partitions.
>>
>> As Akka Streams builds on top of the consumer it could implement a
>> similar pattern. Of course, one cannot use `auto.commit` on the
>> consumer, but commits need to be manged manually, (ie, only data that
>> was take out of the buffer and actually was processed can be committed).
>>
>> For the `MessageChooser` idea, I also still think it might be useful,
>> but it's unclear to me if this should be a consumer feature or build on
>> top of the consumer (maybe it could be a Streams feature, as Streams is
>> build on top of the consumer). Thoughts?
>>
>>
>> -Matthias
>>
>> On 10/18/18 9:27 AM, Jan Filipiak wrote:
>>> The idea for you would be that Messagechooser could hang on to the
>>> prefetched messages.
>>>
>>> ccing cmccabe@apache.org
>>>
>>> @Collin
>>> just for you to see that MessageChooser is a powerfull abstraction.
>>>
>>> :)
>>>
>>> Best jan
>>>
>>> On 18.10.2018 13:59, Zahari Dichev wrote:
>>>> Jan,
>>>>
>>>> Quite insightful indeed. I think your propositions are valid.
>>>>
>>>> Ryanne,
>>>>
>>>> I understand that consumers are using a pull model... And yes, indeed
>> if a
>>>> consumer is not ready for more records it surely should not call poll.
>>>> Except that it needs to do so periodically in order to indicate that its
>>>> live. Forget about the "backpressure", I guess I was wrong with phrasing
>>>> this so lets not get caught up on it.
>>>>
>>>> You say pause/resume can be used to prioritise certain topics/partitions
>>>> over others. And indeed this is the case. So instead of thinking about
>> it
>>>> in terms of backpressure, lets put it in a different way. The Akka
>> streams
>>>> connector would like to prioritise certain topics over others, using
>> once
>>>> consumer instance. On top of that, add the detail that the priorities
>>>> change quite frequently (which translates to calling pause/resume
>>>> frequently). So all that being said, what would be a proper way to
>> handle
>>>> the situation without throwing the pre-fetched records away when calling
>>>> poll on a consumer that happens to have a topic that was recently paused
>>>> (and that might be un-paused soon )? Am I the only one who considers
>> that
>>>> an actual problem with the use os pause/resume ? Not sure how to explain
>>>> the situation in a better way..
>>>>
>>>> Zahari
>>>>
>>>>
>>>> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <za...@gmail.com>
>>>> wrote:
>>>>
>>>>> Thanks a lot Jan,
>>>>>
>>>>> I will read it.
>>>>>
>>>>> Zahari
>>>>>
>>>>> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Jan.Filipiak@trivago.com
>>>
>>>>> wrote:
>>>>>
>>>>>> especially my suggestions ;)
>>>>>>
>>>>>> On 18.10.2018 08:30, Jan Filipiak wrote:
>>>>>>> Hi Zahari,
>>>>>>>
>>>>>>> would you be willing to scan through the KIP-349 discussion a little?
>>>>>>> I think it has suggestions that could be interesting for you
>>>>>>>
>>>>>>> Best Jan
>>>>>>>
>>>>>>> On 16.10.2018 09:29, Zahari Dichev wrote:
>>>>>>>> Hi there Kafka developers,
>>>>>>>>
>>>>>>>> I am currently trying to find a solution to an issue that has been
>>>>>>>> manifesting itself in the Akka streams implementation of the Kafka
>>>>>>>> connector. When it comes to consuming messages, the implementation
>>>>>> relies
>>>>>>>> heavily on the fact that we can pause and resume partitions. In some
>>>>>>>> situations when a single consumer instance is shared among several
>>>>>>>> streams,
>>>>>>>> we might end up with frequently pausing and unpausing a set of topic
>>>>>>>> partitions, which is the main facility that allows us to implement
>> back
>>>>>>>> pressure. This however has certain disadvantages, especially when
>>>>>>>> there are
>>>>>>>> two consumers that differ in terms of processing speed.
>>>>>>>>
>>>>>>>> To articulate the issue more clearly, imagine that a consumer
>> maintains
>>>>>>>> assignments for two topic partitions *TP1* and *TP2*. This consumer
>> is
>>>>>>>> shared by two streams - S1 and S2. So effectively when we have
>> demand
>>>>>>>> from
>>>>>>>> only one of the streams - *S1*, we will pause one of the topic
>>>>>> partitions
>>>>>>>> *TP2* and call *poll()* on the consumer to only retrieve the records
>>>>>> for
>>>>>>>> the demanded topic partition - *TP1*. The result of that is all the
>>>>>>>> records
>>>>>>>> that have been prefetched for *TP2* are now thrown away by the
>> fetcher
>>>>>>>> ("*Not
>>>>>>>> returning fetched records for assigned partition TP2 since it is no
>>>>>>>> longer
>>>>>>>> fetchable"*). If we extrapolate that to multiple streams sharing the
>>>>>> same
>>>>>>>> consumer, we might quickly end up in a situation where we throw
>>>>>>>> prefetched
>>>>>>>> data quite often. This does not seem like the most efficient
>> approach
>>>>>> and
>>>>>>>> in fact produces quite a lot of overlapping fetch requests as
>>>>>> illustrated
>>>>>>>> in the following issue:
>>>>>>>>
>>>>>>>> https://github.com/akka/alpakka-kafka/issues/549
>>>>>>>>
>>>>>>>> I am writing this email to get some initial opinion on a KIP I was
>>>>>>>> thinking
>>>>>>>> about. What if we give the clients of the Consumer API a bit more
>>>>>> control
>>>>>>>> of what to do with this prefetched data. Two options I am wondering
>>>>>>>> about:
>>>>>>>>
>>>>>>>> 1. Introduce a configuration setting, such as*
>>>>>>>> "return-prefetched-data-for-paused-topic-partitions = false"* (have
>> to
>>>>>>>> think of a better name), which when set to true will return what is
>>>>>>>> prefetched instead of throwing it away on calling *poll()*. Since
>> this
>>>>>> is
>>>>>>>> amount of data that is bounded by the maximum size of the prefetch,
>> we
>>>>>>>> can
>>>>>>>> control what is the most amount of records returned. The client of
>> the
>>>>>>>> consumer API can then be responsible for keeping that data around
>> and
>>>>>> use
>>>>>>>> it when appropriate (i.e. when demand is present)
>>>>>>>>
>>>>>>>> 2. Introduce a facility to pass in a buffer into which the
>> prefetched
>>>>>>>> records are drained when poll is called and paused partitions have
>> some
>>>>>>>> prefetched records.
>>>>>>>>
>>>>>>>> Any opinions on the matter are welcome. Thanks a lot !
>>>>>>>>
>>>>>>>> Zahari Dichev
>>>>>>>>
>>>>>>
>>>>>
>>>>
>>
>>
> 

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Hi there Matthias,

Very useful thoughts indeed. I have considered the exact same approach but
what worries me a bit is that I do not think that will certainly solve the
issue. Imagine the following situation. You have one consumer that is quite
slow so lets say it call poll every 5 seconds, while you need to call poll
every 1 second to issue a heartbeat (these are made up numbers of course).

So our consumer calls poll at T0 grabs some data and puts it in a buffer
and calls pause on the topic partition. The fetcher tries to pipeline and
issues a fetch request and at some point the data arrives. At that point we
have some data in the buffer and we can do whatever we want with it, but
there is also some data living within the consumer/fetcher. Approximately
one second later we call poll again because we need to. We are not getting
any data because our partition is paused and this is good because if we got
data we would not know what to do with it as our client is still busy
crunching the data from the first poll. So far so good. What happens though
is that the pre-fetched data gets thrown away upon calling poll as its "no
longer fetchable...".  Maybe I am not fully understanding your suggested
approach, but I dont think it would solve this problem.

Zahari

On Fri, Oct 19, 2018 at 8:10 PM Matthias J. Sax <ma...@confluent.io>
wrote:

> Just my 2 cents.
>
> I am not 100% sure if we would need to change the consumer for this.
> While I still think, that KIP-349 might be valuable, it seems to be
> complementary/orthogonal to the issue discussed here.
>
> For Kafka Streams, we have a related scenario and what Kafka Streams
> does is, to add its own buffer on top of the consumer. Thus, for each
> `poll()` all data is put into this buffer, and now Streams can decide
> which record to process first. For buffers that have data, we can call
> `pause()` without loosing fetched data (not 100% percent guarantee,
> depending on fetch size and max.poll.record etc, but with high
> probability) and if a buffer gets empty we `resume()` partitions.
>
> As Akka Streams builds on top of the consumer it could implement a
> similar pattern. Of course, one cannot use `auto.commit` on the
> consumer, but commits need to be manged manually, (ie, only data that
> was take out of the buffer and actually was processed can be committed).
>
> For the `MessageChooser` idea, I also still think it might be useful,
> but it's unclear to me if this should be a consumer feature or build on
> top of the consumer (maybe it could be a Streams feature, as Streams is
> build on top of the consumer). Thoughts?
>
>
> -Matthias
>
> On 10/18/18 9:27 AM, Jan Filipiak wrote:
> > The idea for you would be that Messagechooser could hang on to the
> > prefetched messages.
> >
> > ccing cmccabe@apache.org
> >
> > @Collin
> > just for you to see that MessageChooser is a powerfull abstraction.
> >
> > :)
> >
> > Best jan
> >
> > On 18.10.2018 13:59, Zahari Dichev wrote:
> >> Jan,
> >>
> >> Quite insightful indeed. I think your propositions are valid.
> >>
> >> Ryanne,
> >>
> >> I understand that consumers are using a pull model... And yes, indeed
> if a
> >> consumer is not ready for more records it surely should not call poll.
> >> Except that it needs to do so periodically in order to indicate that its
> >> live. Forget about the "backpressure", I guess I was wrong with phrasing
> >> this so lets not get caught up on it.
> >>
> >> You say pause/resume can be used to prioritise certain topics/partitions
> >> over others. And indeed this is the case. So instead of thinking about
> it
> >> in terms of backpressure, lets put it in a different way. The Akka
> streams
> >> connector would like to prioritise certain topics over others, using
> once
> >> consumer instance. On top of that, add the detail that the priorities
> >> change quite frequently (which translates to calling pause/resume
> >> frequently). So all that being said, what would be a proper way to
> handle
> >> the situation without throwing the pre-fetched records away when calling
> >> poll on a consumer that happens to have a topic that was recently paused
> >> (and that might be un-paused soon )? Am I the only one who considers
> that
> >> an actual problem with the use os pause/resume ? Not sure how to explain
> >> the situation in a better way..
> >>
> >> Zahari
> >>
> >>
> >> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <za...@gmail.com>
> >> wrote:
> >>
> >>> Thanks a lot Jan,
> >>>
> >>> I will read it.
> >>>
> >>> Zahari
> >>>
> >>> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Jan.Filipiak@trivago.com
> >
> >>> wrote:
> >>>
> >>>> especially my suggestions ;)
> >>>>
> >>>> On 18.10.2018 08:30, Jan Filipiak wrote:
> >>>>> Hi Zahari,
> >>>>>
> >>>>> would you be willing to scan through the KIP-349 discussion a little?
> >>>>> I think it has suggestions that could be interesting for you
> >>>>>
> >>>>> Best Jan
> >>>>>
> >>>>> On 16.10.2018 09:29, Zahari Dichev wrote:
> >>>>>> Hi there Kafka developers,
> >>>>>>
> >>>>>> I am currently trying to find a solution to an issue that has been
> >>>>>> manifesting itself in the Akka streams implementation of the Kafka
> >>>>>> connector. When it comes to consuming messages, the implementation
> >>>> relies
> >>>>>> heavily on the fact that we can pause and resume partitions. In some
> >>>>>> situations when a single consumer instance is shared among several
> >>>>>> streams,
> >>>>>> we might end up with frequently pausing and unpausing a set of topic
> >>>>>> partitions, which is the main facility that allows us to implement
> back
> >>>>>> pressure. This however has certain disadvantages, especially when
> >>>>>> there are
> >>>>>> two consumers that differ in terms of processing speed.
> >>>>>>
> >>>>>> To articulate the issue more clearly, imagine that a consumer
> maintains
> >>>>>> assignments for two topic partitions *TP1* and *TP2*. This consumer
> is
> >>>>>> shared by two streams - S1 and S2. So effectively when we have
> demand
> >>>>>> from
> >>>>>> only one of the streams - *S1*, we will pause one of the topic
> >>>> partitions
> >>>>>> *TP2* and call *poll()* on the consumer to only retrieve the records
> >>>> for
> >>>>>> the demanded topic partition - *TP1*. The result of that is all the
> >>>>>> records
> >>>>>> that have been prefetched for *TP2* are now thrown away by the
> fetcher
> >>>>>> ("*Not
> >>>>>> returning fetched records for assigned partition TP2 since it is no
> >>>>>> longer
> >>>>>> fetchable"*). If we extrapolate that to multiple streams sharing the
> >>>> same
> >>>>>> consumer, we might quickly end up in a situation where we throw
> >>>>>> prefetched
> >>>>>> data quite often. This does not seem like the most efficient
> approach
> >>>> and
> >>>>>> in fact produces quite a lot of overlapping fetch requests as
> >>>> illustrated
> >>>>>> in the following issue:
> >>>>>>
> >>>>>> https://github.com/akka/alpakka-kafka/issues/549
> >>>>>>
> >>>>>> I am writing this email to get some initial opinion on a KIP I was
> >>>>>> thinking
> >>>>>> about. What if we give the clients of the Consumer API a bit more
> >>>> control
> >>>>>> of what to do with this prefetched data. Two options I am wondering
> >>>>>> about:
> >>>>>>
> >>>>>> 1. Introduce a configuration setting, such as*
> >>>>>> "return-prefetched-data-for-paused-topic-partitions = false"* (have
> to
> >>>>>> think of a better name), which when set to true will return what is
> >>>>>> prefetched instead of throwing it away on calling *poll()*. Since
> this
> >>>> is
> >>>>>> amount of data that is bounded by the maximum size of the prefetch,
> we
> >>>>>> can
> >>>>>> control what is the most amount of records returned. The client of
> the
> >>>>>> consumer API can then be responsible for keeping that data around
> and
> >>>> use
> >>>>>> it when appropriate (i.e. when demand is present)
> >>>>>>
> >>>>>> 2. Introduce a facility to pass in a buffer into which the
> prefetched
> >>>>>> records are drained when poll is called and paused partitions have
> some
> >>>>>> prefetched records.
> >>>>>>
> >>>>>> Any opinions on the matter are welcome. Thanks a lot !
> >>>>>>
> >>>>>> Zahari Dichev
> >>>>>>
> >>>>
> >>>
> >>
>
>

Re: Throwing away prefetched records optimisation.

["Matthias J. Sax" <ma...@confluent.io>]

Just my 2 cents.

I am not 100% sure if we would need to change the consumer for this.
While I still think, that KIP-349 might be valuable, it seems to be
complementary/orthogonal to the issue discussed here.

For Kafka Streams, we have a related scenario and what Kafka Streams
does is, to add its own buffer on top of the consumer. Thus, for each
`poll()` all data is put into this buffer, and now Streams can decide
which record to process first. For buffers that have data, we can call
`pause()` without loosing fetched data (not 100% percent guarantee,
depending on fetch size and max.poll.record etc, but with high
probability) and if a buffer gets empty we `resume()` partitions.

As Akka Streams builds on top of the consumer it could implement a
similar pattern. Of course, one cannot use `auto.commit` on the
consumer, but commits need to be manged manually, (ie, only data that
was take out of the buffer and actually was processed can be committed).

For the `MessageChooser` idea, I also still think it might be useful,
but it's unclear to me if this should be a consumer feature or build on
top of the consumer (maybe it could be a Streams feature, as Streams is
build on top of the consumer). Thoughts?


-Matthias

On 10/18/18 9:27 AM, Jan Filipiak wrote:
> The idea for you would be that Messagechooser could hang on to the 
> prefetched messages.
> 
> ccing cmccabe@apache.org
> 
> @Collin
> just for you to see that MessageChooser is a powerfull abstraction.
> 
> :)
> 
> Best jan
> 
> On 18.10.2018 13:59, Zahari Dichev wrote:
>> Jan,
>>
>> Quite insightful indeed. I think your propositions are valid.
>>
>> Ryanne,
>>
>> I understand that consumers are using a pull model... And yes, indeed if a
>> consumer is not ready for more records it surely should not call poll.
>> Except that it needs to do so periodically in order to indicate that its
>> live. Forget about the "backpressure", I guess I was wrong with phrasing
>> this so lets not get caught up on it.
>>
>> You say pause/resume can be used to prioritise certain topics/partitions
>> over others. And indeed this is the case. So instead of thinking about it
>> in terms of backpressure, lets put it in a different way. The Akka streams
>> connector would like to prioritise certain topics over others, using once
>> consumer instance. On top of that, add the detail that the priorities
>> change quite frequently (which translates to calling pause/resume
>> frequently). So all that being said, what would be a proper way to handle
>> the situation without throwing the pre-fetched records away when calling
>> poll on a consumer that happens to have a topic that was recently paused
>> (and that might be un-paused soon )? Am I the only one who considers that
>> an actual problem with the use os pause/resume ? Not sure how to explain
>> the situation in a better way..
>>
>> Zahari
>>
>>
>> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <za...@gmail.com>
>> wrote:
>>
>>> Thanks a lot Jan,
>>>
>>> I will read it.
>>>
>>> Zahari
>>>
>>> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Ja...@trivago.com>
>>> wrote:
>>>
>>>> especially my suggestions ;)
>>>>
>>>> On 18.10.2018 08:30, Jan Filipiak wrote:
>>>>> Hi Zahari,
>>>>>
>>>>> would you be willing to scan through the KIP-349 discussion a little?
>>>>> I think it has suggestions that could be interesting for you
>>>>>
>>>>> Best Jan
>>>>>
>>>>> On 16.10.2018 09:29, Zahari Dichev wrote:
>>>>>> Hi there Kafka developers,
>>>>>>
>>>>>> I am currently trying to find a solution to an issue that has been
>>>>>> manifesting itself in the Akka streams implementation of the Kafka
>>>>>> connector. When it comes to consuming messages, the implementation
>>>> relies
>>>>>> heavily on the fact that we can pause and resume partitions. In some
>>>>>> situations when a single consumer instance is shared among several
>>>>>> streams,
>>>>>> we might end up with frequently pausing and unpausing a set of topic
>>>>>> partitions, which is the main facility that allows us to implement back
>>>>>> pressure. This however has certain disadvantages, especially when
>>>>>> there are
>>>>>> two consumers that differ in terms of processing speed.
>>>>>>
>>>>>> To articulate the issue more clearly, imagine that a consumer maintains
>>>>>> assignments for two topic partitions *TP1* and *TP2*. This consumer is
>>>>>> shared by two streams - S1 and S2. So effectively when we have demand
>>>>>> from
>>>>>> only one of the streams - *S1*, we will pause one of the topic
>>>> partitions
>>>>>> *TP2* and call *poll()* on the consumer to only retrieve the records
>>>> for
>>>>>> the demanded topic partition - *TP1*. The result of that is all the
>>>>>> records
>>>>>> that have been prefetched for *TP2* are now thrown away by the fetcher
>>>>>> ("*Not
>>>>>> returning fetched records for assigned partition TP2 since it is no
>>>>>> longer
>>>>>> fetchable"*). If we extrapolate that to multiple streams sharing the
>>>> same
>>>>>> consumer, we might quickly end up in a situation where we throw
>>>>>> prefetched
>>>>>> data quite often. This does not seem like the most efficient approach
>>>> and
>>>>>> in fact produces quite a lot of overlapping fetch requests as
>>>> illustrated
>>>>>> in the following issue:
>>>>>>
>>>>>> https://github.com/akka/alpakka-kafka/issues/549
>>>>>>
>>>>>> I am writing this email to get some initial opinion on a KIP I was
>>>>>> thinking
>>>>>> about. What if we give the clients of the Consumer API a bit more
>>>> control
>>>>>> of what to do with this prefetched data. Two options I am wondering
>>>>>> about:
>>>>>>
>>>>>> 1. Introduce a configuration setting, such as*
>>>>>> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
>>>>>> think of a better name), which when set to true will return what is
>>>>>> prefetched instead of throwing it away on calling *poll()*. Since this
>>>> is
>>>>>> amount of data that is bounded by the maximum size of the prefetch, we
>>>>>> can
>>>>>> control what is the most amount of records returned. The client of the
>>>>>> consumer API can then be responsible for keeping that data around and
>>>> use
>>>>>> it when appropriate (i.e. when demand is present)
>>>>>>
>>>>>> 2. Introduce a facility to pass in a buffer into which the prefetched
>>>>>> records are drained when poll is called and paused partitions have some
>>>>>> prefetched records.
>>>>>>
>>>>>> Any opinions on the matter are welcome. Thanks a lot !
>>>>>>
>>>>>> Zahari Dichev
>>>>>>
>>>>
>>>
>>

Re: Throwing away prefetched records optimisation.

[Jan Filipiak <Ja...@trivago.com>]

The idea for you would be that Messagechooser could hang on to the 
prefetched messages.

ccing cmccabe@apache.org

@Collin
just for you to see that MessageChooser is a powerfull abstraction.

:)

Best jan

On 18.10.2018 13:59, Zahari Dichev wrote:
> Jan,
>
> Quite insightful indeed. I think your propositions are valid.
>
> Ryanne,
>
> I understand that consumers are using a pull model... And yes, indeed if a
> consumer is not ready for more records it surely should not call poll.
> Except that it needs to do so periodically in order to indicate that its
> live. Forget about the "backpressure", I guess I was wrong with phrasing
> this so lets not get caught up on it.
>
> You say pause/resume can be used to prioritise certain topics/partitions
> over others. And indeed this is the case. So instead of thinking about it
> in terms of backpressure, lets put it in a different way. The Akka streams
> connector would like to prioritise certain topics over others, using once
> consumer instance. On top of that, add the detail that the priorities
> change quite frequently (which translates to calling pause/resume
> frequently). So all that being said, what would be a proper way to handle
> the situation without throwing the pre-fetched records away when calling
> poll on a consumer that happens to have a topic that was recently paused
> (and that might be un-paused soon )? Am I the only one who considers that
> an actual problem with the use os pause/resume ? Not sure how to explain
> the situation in a better way..
>
> Zahari
>
>
> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <za...@gmail.com>
> wrote:
>
>> Thanks a lot Jan,
>>
>> I will read it.
>>
>> Zahari
>>
>> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Ja...@trivago.com>
>> wrote:
>>
>>> especially my suggestions ;)
>>>
>>> On 18.10.2018 08:30, Jan Filipiak wrote:
>>>> Hi Zahari,
>>>>
>>>> would you be willing to scan through the KIP-349 discussion a little?
>>>> I think it has suggestions that could be interesting for you
>>>>
>>>> Best Jan
>>>>
>>>> On 16.10.2018 09:29, Zahari Dichev wrote:
>>>>> Hi there Kafka developers,
>>>>>
>>>>> I am currently trying to find a solution to an issue that has been
>>>>> manifesting itself in the Akka streams implementation of the Kafka
>>>>> connector. When it comes to consuming messages, the implementation
>>> relies
>>>>> heavily on the fact that we can pause and resume partitions. In some
>>>>> situations when a single consumer instance is shared among several
>>>>> streams,
>>>>> we might end up with frequently pausing and unpausing a set of topic
>>>>> partitions, which is the main facility that allows us to implement back
>>>>> pressure. This however has certain disadvantages, especially when
>>>>> there are
>>>>> two consumers that differ in terms of processing speed.
>>>>>
>>>>> To articulate the issue more clearly, imagine that a consumer maintains
>>>>> assignments for two topic partitions *TP1* and *TP2*. This consumer is
>>>>> shared by two streams - S1 and S2. So effectively when we have demand
>>>>> from
>>>>> only one of the streams - *S1*, we will pause one of the topic
>>> partitions
>>>>> *TP2* and call *poll()* on the consumer to only retrieve the records
>>> for
>>>>> the demanded topic partition - *TP1*. The result of that is all the
>>>>> records
>>>>> that have been prefetched for *TP2* are now thrown away by the fetcher
>>>>> ("*Not
>>>>> returning fetched records for assigned partition TP2 since it is no
>>>>> longer
>>>>> fetchable"*). If we extrapolate that to multiple streams sharing the
>>> same
>>>>> consumer, we might quickly end up in a situation where we throw
>>>>> prefetched
>>>>> data quite often. This does not seem like the most efficient approach
>>> and
>>>>> in fact produces quite a lot of overlapping fetch requests as
>>> illustrated
>>>>> in the following issue:
>>>>>
>>>>> https://github.com/akka/alpakka-kafka/issues/549
>>>>>
>>>>> I am writing this email to get some initial opinion on a KIP I was
>>>>> thinking
>>>>> about. What if we give the clients of the Consumer API a bit more
>>> control
>>>>> of what to do with this prefetched data. Two options I am wondering
>>>>> about:
>>>>>
>>>>> 1. Introduce a configuration setting, such as*
>>>>> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
>>>>> think of a better name), which when set to true will return what is
>>>>> prefetched instead of throwing it away on calling *poll()*. Since this
>>> is
>>>>> amount of data that is bounded by the maximum size of the prefetch, we
>>>>> can
>>>>> control what is the most amount of records returned. The client of the
>>>>> consumer API can then be responsible for keeping that data around and
>>> use
>>>>> it when appropriate (i.e. when demand is present)
>>>>>
>>>>> 2. Introduce a facility to pass in a buffer into which the prefetched
>>>>> records are drained when poll is called and paused partitions have some
>>>>> prefetched records.
>>>>>
>>>>> Any opinions on the matter are welcome. Thanks a lot !
>>>>>
>>>>> Zahari Dichev
>>>>>
>>>
>>
>

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Thanks Ryanne,
I am glad it makes sense. Should I put a KIP together and call for
discussion on it ? Its my first KIP, so have not quite locked in the
process yet.

Zahari

On Thu, Oct 18, 2018 at 3:26 PM Ryanne Dolan <ry...@gmail.com> wrote:

> Zahari, that makes sense, thanks for reframing your question. I suspect
> that pause/resume was not intended to be called at high frequency like
> that, but I agree with you that the current behavior is needlessly
> inefficient. I like your idea of making it configurable.
>
> Ryanne
>
> On Thu, Oct 18, 2018, 6:59 AM Zahari Dichev <za...@gmail.com>
> wrote:
>
> > Jan,
> >
> > Quite insightful indeed. I think your propositions are valid.
> >
> > Ryanne,
> >
> > I understand that consumers are using a pull model... And yes, indeed if
> a
> > consumer is not ready for more records it surely should not call poll.
> > Except that it needs to do so periodically in order to indicate that its
> > live. Forget about the "backpressure", I guess I was wrong with phrasing
> > this so lets not get caught up on it.
> >
> > You say pause/resume can be used to prioritise certain topics/partitions
> > over others. And indeed this is the case. So instead of thinking about it
> > in terms of backpressure, lets put it in a different way. The Akka
> streams
> > connector would like to prioritise certain topics over others, using once
> > consumer instance. On top of that, add the detail that the priorities
> > change quite frequently (which translates to calling pause/resume
> > frequently). So all that being said, what would be a proper way to handle
> > the situation without throwing the pre-fetched records away when calling
> > poll on a consumer that happens to have a topic that was recently paused
> > (and that might be un-paused soon )? Am I the only one who considers that
> > an actual problem with the use os pause/resume ? Not sure how to explain
> > the situation in a better way..
> >
> > Zahari
> >
> >
> > On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <za...@gmail.com>
> > wrote:
> >
> > > Thanks a lot Jan,
> > >
> > > I will read it.
> > >
> > > Zahari
> > >
> > > On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Jan.Filipiak@trivago.com
> >
> > > wrote:
> > >
> > >> especially my suggestions ;)
> > >>
> > >> On 18.10.2018 08:30, Jan Filipiak wrote:
> > >> > Hi Zahari,
> > >> >
> > >> > would you be willing to scan through the KIP-349 discussion a
> little?
> > >> > I think it has suggestions that could be interesting for you
> > >> >
> > >> > Best Jan
> > >> >
> > >> > On 16.10.2018 09:29, Zahari Dichev wrote:
> > >> >> Hi there Kafka developers,
> > >> >>
> > >> >> I am currently trying to find a solution to an issue that has been
> > >> >> manifesting itself in the Akka streams implementation of the Kafka
> > >> >> connector. When it comes to consuming messages, the implementation
> > >> relies
> > >> >> heavily on the fact that we can pause and resume partitions. In
> some
> > >> >> situations when a single consumer instance is shared among several
> > >> >> streams,
> > >> >> we might end up with frequently pausing and unpausing a set of
> topic
> > >> >> partitions, which is the main facility that allows us to implement
> > back
> > >> >> pressure. This however has certain disadvantages, especially when
> > >> >> there are
> > >> >> two consumers that differ in terms of processing speed.
> > >> >>
> > >> >> To articulate the issue more clearly, imagine that a consumer
> > maintains
> > >> >> assignments for two topic partitions *TP1* and *TP2*. This consumer
> > is
> > >> >> shared by two streams - S1 and S2. So effectively when we have
> demand
> > >> >> from
> > >> >> only one of the streams - *S1*, we will pause one of the topic
> > >> partitions
> > >> >> *TP2* and call *poll()* on the consumer to only retrieve the
> records
> > >> for
> > >> >> the demanded topic partition - *TP1*. The result of that is all the
> > >> >> records
> > >> >> that have been prefetched for *TP2* are now thrown away by the
> > fetcher
> > >> >> ("*Not
> > >> >> returning fetched records for assigned partition TP2 since it is no
> > >> >> longer
> > >> >> fetchable"*). If we extrapolate that to multiple streams sharing
> the
> > >> same
> > >> >> consumer, we might quickly end up in a situation where we throw
> > >> >> prefetched
> > >> >> data quite often. This does not seem like the most efficient
> approach
> > >> and
> > >> >> in fact produces quite a lot of overlapping fetch requests as
> > >> illustrated
> > >> >> in the following issue:
> > >> >>
> > >> >> https://github.com/akka/alpakka-kafka/issues/549
> > >> >>
> > >> >> I am writing this email to get some initial opinion on a KIP I was
> > >> >> thinking
> > >> >> about. What if we give the clients of the Consumer API a bit more
> > >> control
> > >> >> of what to do with this prefetched data. Two options I am wondering
> > >> >> about:
> > >> >>
> > >> >> 1. Introduce a configuration setting, such as*
> > >> >> "return-prefetched-data-for-paused-topic-partitions = false"* (have
> > to
> > >> >> think of a better name), which when set to true will return what is
> > >> >> prefetched instead of throwing it away on calling *poll()*. Since
> > this
> > >> is
> > >> >> amount of data that is bounded by the maximum size of the prefetch,
> > we
> > >> >> can
> > >> >> control what is the most amount of records returned. The client of
> > the
> > >> >> consumer API can then be responsible for keeping that data around
> and
> > >> use
> > >> >> it when appropriate (i.e. when demand is present)
> > >> >>
> > >> >> 2. Introduce a facility to pass in a buffer into which the
> prefetched
> > >> >> records are drained when poll is called and paused partitions have
> > some
> > >> >> prefetched records.
> > >> >>
> > >> >> Any opinions on the matter are welcome. Thanks a lot !
> > >> >>
> > >> >> Zahari Dichev
> > >> >>
> > >>
> > >
> >
>

Re: Throwing away prefetched records optimisation.

[Ryanne Dolan <ry...@gmail.com>]

Zahari, that makes sense, thanks for reframing your question. I suspect
that pause/resume was not intended to be called at high frequency like
that, but I agree with you that the current behavior is needlessly
inefficient. I like your idea of making it configurable.

Ryanne

On Thu, Oct 18, 2018, 6:59 AM Zahari Dichev <za...@gmail.com> wrote:

> Jan,
>
> Quite insightful indeed. I think your propositions are valid.
>
> Ryanne,
>
> I understand that consumers are using a pull model... And yes, indeed if a
> consumer is not ready for more records it surely should not call poll.
> Except that it needs to do so periodically in order to indicate that its
> live. Forget about the "backpressure", I guess I was wrong with phrasing
> this so lets not get caught up on it.
>
> You say pause/resume can be used to prioritise certain topics/partitions
> over others. And indeed this is the case. So instead of thinking about it
> in terms of backpressure, lets put it in a different way. The Akka streams
> connector would like to prioritise certain topics over others, using once
> consumer instance. On top of that, add the detail that the priorities
> change quite frequently (which translates to calling pause/resume
> frequently). So all that being said, what would be a proper way to handle
> the situation without throwing the pre-fetched records away when calling
> poll on a consumer that happens to have a topic that was recently paused
> (and that might be un-paused soon )? Am I the only one who considers that
> an actual problem with the use os pause/resume ? Not sure how to explain
> the situation in a better way..
>
> Zahari
>
>
> On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <za...@gmail.com>
> wrote:
>
> > Thanks a lot Jan,
> >
> > I will read it.
> >
> > Zahari
> >
> > On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Ja...@trivago.com>
> > wrote:
> >
> >> especially my suggestions ;)
> >>
> >> On 18.10.2018 08:30, Jan Filipiak wrote:
> >> > Hi Zahari,
> >> >
> >> > would you be willing to scan through the KIP-349 discussion a little?
> >> > I think it has suggestions that could be interesting for you
> >> >
> >> > Best Jan
> >> >
> >> > On 16.10.2018 09:29, Zahari Dichev wrote:
> >> >> Hi there Kafka developers,
> >> >>
> >> >> I am currently trying to find a solution to an issue that has been
> >> >> manifesting itself in the Akka streams implementation of the Kafka
> >> >> connector. When it comes to consuming messages, the implementation
> >> relies
> >> >> heavily on the fact that we can pause and resume partitions. In some
> >> >> situations when a single consumer instance is shared among several
> >> >> streams,
> >> >> we might end up with frequently pausing and unpausing a set of topic
> >> >> partitions, which is the main facility that allows us to implement
> back
> >> >> pressure. This however has certain disadvantages, especially when
> >> >> there are
> >> >> two consumers that differ in terms of processing speed.
> >> >>
> >> >> To articulate the issue more clearly, imagine that a consumer
> maintains
> >> >> assignments for two topic partitions *TP1* and *TP2*. This consumer
> is
> >> >> shared by two streams - S1 and S2. So effectively when we have demand
> >> >> from
> >> >> only one of the streams - *S1*, we will pause one of the topic
> >> partitions
> >> >> *TP2* and call *poll()* on the consumer to only retrieve the records
> >> for
> >> >> the demanded topic partition - *TP1*. The result of that is all the
> >> >> records
> >> >> that have been prefetched for *TP2* are now thrown away by the
> fetcher
> >> >> ("*Not
> >> >> returning fetched records for assigned partition TP2 since it is no
> >> >> longer
> >> >> fetchable"*). If we extrapolate that to multiple streams sharing the
> >> same
> >> >> consumer, we might quickly end up in a situation where we throw
> >> >> prefetched
> >> >> data quite often. This does not seem like the most efficient approach
> >> and
> >> >> in fact produces quite a lot of overlapping fetch requests as
> >> illustrated
> >> >> in the following issue:
> >> >>
> >> >> https://github.com/akka/alpakka-kafka/issues/549
> >> >>
> >> >> I am writing this email to get some initial opinion on a KIP I was
> >> >> thinking
> >> >> about. What if we give the clients of the Consumer API a bit more
> >> control
> >> >> of what to do with this prefetched data. Two options I am wondering
> >> >> about:
> >> >>
> >> >> 1. Introduce a configuration setting, such as*
> >> >> "return-prefetched-data-for-paused-topic-partitions = false"* (have
> to
> >> >> think of a better name), which when set to true will return what is
> >> >> prefetched instead of throwing it away on calling *poll()*. Since
> this
> >> is
> >> >> amount of data that is bounded by the maximum size of the prefetch,
> we
> >> >> can
> >> >> control what is the most amount of records returned. The client of
> the
> >> >> consumer API can then be responsible for keeping that data around and
> >> use
> >> >> it when appropriate (i.e. when demand is present)
> >> >>
> >> >> 2. Introduce a facility to pass in a buffer into which the prefetched
> >> >> records are drained when poll is called and paused partitions have
> some
> >> >> prefetched records.
> >> >>
> >> >> Any opinions on the matter are welcome. Thanks a lot !
> >> >>
> >> >> Zahari Dichev
> >> >>
> >>
> >
>

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Jan,

Quite insightful indeed. I think your propositions are valid.

Ryanne,

I understand that consumers are using a pull model... And yes, indeed if a
consumer is not ready for more records it surely should not call poll.
Except that it needs to do so periodically in order to indicate that its
live. Forget about the "backpressure", I guess I was wrong with phrasing
this so lets not get caught up on it.

You say pause/resume can be used to prioritise certain topics/partitions
over others. And indeed this is the case. So instead of thinking about it
in terms of backpressure, lets put it in a different way. The Akka streams
connector would like to prioritise certain topics over others, using once
consumer instance. On top of that, add the detail that the priorities
change quite frequently (which translates to calling pause/resume
frequently). So all that being said, what would be a proper way to handle
the situation without throwing the pre-fetched records away when calling
poll on a consumer that happens to have a topic that was recently paused
(and that might be un-paused soon )? Am I the only one who considers that
an actual problem with the use os pause/resume ? Not sure how to explain
the situation in a better way..

Zahari


On Thu, Oct 18, 2018 at 9:46 AM Zahari Dichev <za...@gmail.com>
wrote:

> Thanks a lot Jan,
>
> I will read it.
>
> Zahari
>
> On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Ja...@trivago.com>
> wrote:
>
>> especially my suggestions ;)
>>
>> On 18.10.2018 08:30, Jan Filipiak wrote:
>> > Hi Zahari,
>> >
>> > would you be willing to scan through the KIP-349 discussion a little?
>> > I think it has suggestions that could be interesting for you
>> >
>> > Best Jan
>> >
>> > On 16.10.2018 09:29, Zahari Dichev wrote:
>> >> Hi there Kafka developers,
>> >>
>> >> I am currently trying to find a solution to an issue that has been
>> >> manifesting itself in the Akka streams implementation of the Kafka
>> >> connector. When it comes to consuming messages, the implementation
>> relies
>> >> heavily on the fact that we can pause and resume partitions. In some
>> >> situations when a single consumer instance is shared among several
>> >> streams,
>> >> we might end up with frequently pausing and unpausing a set of topic
>> >> partitions, which is the main facility that allows us to implement back
>> >> pressure. This however has certain disadvantages, especially when
>> >> there are
>> >> two consumers that differ in terms of processing speed.
>> >>
>> >> To articulate the issue more clearly, imagine that a consumer maintains
>> >> assignments for two topic partitions *TP1* and *TP2*. This consumer is
>> >> shared by two streams - S1 and S2. So effectively when we have demand
>> >> from
>> >> only one of the streams - *S1*, we will pause one of the topic
>> partitions
>> >> *TP2* and call *poll()* on the consumer to only retrieve the records
>> for
>> >> the demanded topic partition - *TP1*. The result of that is all the
>> >> records
>> >> that have been prefetched for *TP2* are now thrown away by the fetcher
>> >> ("*Not
>> >> returning fetched records for assigned partition TP2 since it is no
>> >> longer
>> >> fetchable"*). If we extrapolate that to multiple streams sharing the
>> same
>> >> consumer, we might quickly end up in a situation where we throw
>> >> prefetched
>> >> data quite often. This does not seem like the most efficient approach
>> and
>> >> in fact produces quite a lot of overlapping fetch requests as
>> illustrated
>> >> in the following issue:
>> >>
>> >> https://github.com/akka/alpakka-kafka/issues/549
>> >>
>> >> I am writing this email to get some initial opinion on a KIP I was
>> >> thinking
>> >> about. What if we give the clients of the Consumer API a bit more
>> control
>> >> of what to do with this prefetched data. Two options I am wondering
>> >> about:
>> >>
>> >> 1. Introduce a configuration setting, such as*
>> >> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
>> >> think of a better name), which when set to true will return what is
>> >> prefetched instead of throwing it away on calling *poll()*. Since this
>> is
>> >> amount of data that is bounded by the maximum size of the prefetch, we
>> >> can
>> >> control what is the most amount of records returned. The client of the
>> >> consumer API can then be responsible for keeping that data around and
>> use
>> >> it when appropriate (i.e. when demand is present)
>> >>
>> >> 2. Introduce a facility to pass in a buffer into which the prefetched
>> >> records are drained when poll is called and paused partitions have some
>> >> prefetched records.
>> >>
>> >> Any opinions on the matter are welcome. Thanks a lot !
>> >>
>> >> Zahari Dichev
>> >>
>>
>

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Thanks a lot Jan,

I will read it.

Zahari

On Thu, Oct 18, 2018 at 9:31 AM Jan Filipiak <Ja...@trivago.com>
wrote:

> especially my suggestions ;)
>
> On 18.10.2018 08:30, Jan Filipiak wrote:
> > Hi Zahari,
> >
> > would you be willing to scan through the KIP-349 discussion a little?
> > I think it has suggestions that could be interesting for you
> >
> > Best Jan
> >
> > On 16.10.2018 09:29, Zahari Dichev wrote:
> >> Hi there Kafka developers,
> >>
> >> I am currently trying to find a solution to an issue that has been
> >> manifesting itself in the Akka streams implementation of the Kafka
> >> connector. When it comes to consuming messages, the implementation
> relies
> >> heavily on the fact that we can pause and resume partitions. In some
> >> situations when a single consumer instance is shared among several
> >> streams,
> >> we might end up with frequently pausing and unpausing a set of topic
> >> partitions, which is the main facility that allows us to implement back
> >> pressure. This however has certain disadvantages, especially when
> >> there are
> >> two consumers that differ in terms of processing speed.
> >>
> >> To articulate the issue more clearly, imagine that a consumer maintains
> >> assignments for two topic partitions *TP1* and *TP2*. This consumer is
> >> shared by two streams - S1 and S2. So effectively when we have demand
> >> from
> >> only one of the streams - *S1*, we will pause one of the topic
> partitions
> >> *TP2* and call *poll()* on the consumer to only retrieve the records for
> >> the demanded topic partition - *TP1*. The result of that is all the
> >> records
> >> that have been prefetched for *TP2* are now thrown away by the fetcher
> >> ("*Not
> >> returning fetched records for assigned partition TP2 since it is no
> >> longer
> >> fetchable"*). If we extrapolate that to multiple streams sharing the
> same
> >> consumer, we might quickly end up in a situation where we throw
> >> prefetched
> >> data quite often. This does not seem like the most efficient approach
> and
> >> in fact produces quite a lot of overlapping fetch requests as
> illustrated
> >> in the following issue:
> >>
> >> https://github.com/akka/alpakka-kafka/issues/549
> >>
> >> I am writing this email to get some initial opinion on a KIP I was
> >> thinking
> >> about. What if we give the clients of the Consumer API a bit more
> control
> >> of what to do with this prefetched data. Two options I am wondering
> >> about:
> >>
> >> 1. Introduce a configuration setting, such as*
> >> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> >> think of a better name), which when set to true will return what is
> >> prefetched instead of throwing it away on calling *poll()*. Since this
> is
> >> amount of data that is bounded by the maximum size of the prefetch, we
> >> can
> >> control what is the most amount of records returned. The client of the
> >> consumer API can then be responsible for keeping that data around and
> use
> >> it when appropriate (i.e. when demand is present)
> >>
> >> 2. Introduce a facility to pass in a buffer into which the prefetched
> >> records are drained when poll is called and paused partitions have some
> >> prefetched records.
> >>
> >> Any opinions on the matter are welcome. Thanks a lot !
> >>
> >> Zahari Dichev
> >>
>

Re: Throwing away prefetched records optimisation.

[Jan Filipiak <Ja...@trivago.com>]

especially my suggestions ;)

On 18.10.2018 08:30, Jan Filipiak wrote:
> Hi Zahari,
>
> would you be willing to scan through the KIP-349 discussion a little?
> I think it has suggestions that could be interesting for you
>
> Best Jan
>
> On 16.10.2018 09:29, Zahari Dichev wrote:
>> Hi there Kafka developers,
>>
>> I am currently trying to find a solution to an issue that has been
>> manifesting itself in the Akka streams implementation of the Kafka
>> connector. When it comes to consuming messages, the implementation relies
>> heavily on the fact that we can pause and resume partitions. In some
>> situations when a single consumer instance is shared among several
>> streams,
>> we might end up with frequently pausing and unpausing a set of topic
>> partitions, which is the main facility that allows us to implement back
>> pressure. This however has certain disadvantages, especially when
>> there are
>> two consumers that differ in terms of processing speed.
>>
>> To articulate the issue more clearly, imagine that a consumer maintains
>> assignments for two topic partitions *TP1* and *TP2*. This consumer is
>> shared by two streams - S1 and S2. So effectively when we have demand
>> from
>> only one of the streams - *S1*, we will pause one of the topic partitions
>> *TP2* and call *poll()* on the consumer to only retrieve the records for
>> the demanded topic partition - *TP1*. The result of that is all the
>> records
>> that have been prefetched for *TP2* are now thrown away by the fetcher
>> ("*Not
>> returning fetched records for assigned partition TP2 since it is no
>> longer
>> fetchable"*). If we extrapolate that to multiple streams sharing the same
>> consumer, we might quickly end up in a situation where we throw
>> prefetched
>> data quite often. This does not seem like the most efficient approach and
>> in fact produces quite a lot of overlapping fetch requests as illustrated
>> in the following issue:
>>
>> https://github.com/akka/alpakka-kafka/issues/549
>>
>> I am writing this email to get some initial opinion on a KIP I was
>> thinking
>> about. What if we give the clients of the Consumer API a bit more control
>> of what to do with this prefetched data. Two options I am wondering
>> about:
>>
>> 1. Introduce a configuration setting, such as*
>> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
>> think of a better name), which when set to true will return what is
>> prefetched instead of throwing it away on calling *poll()*. Since this is
>> amount of data that is bounded by the maximum size of the prefetch, we
>> can
>> control what is the most amount of records returned. The client of the
>> consumer API can then be responsible for keeping that data around and use
>> it when appropriate (i.e. when demand is present)
>>
>> 2. Introduce a facility to pass in a buffer into which the prefetched
>> records are drained when poll is called and paused partitions have some
>> prefetched records.
>>
>> Any opinions on the matter are welcome. Thanks a lot !
>>
>> Zahari Dichev
>>

Re: Throwing away prefetched records optimisation.

[Jan Filipiak <Ja...@trivago.com>]

Hi Zahari,

would you be willing to scan through the KIP-349 discussion a little?
I think it has suggestions that could be interesting for you

Best Jan

On 16.10.2018 09:29, Zahari Dichev wrote:
> Hi there Kafka developers,
>
> I am currently trying to find a solution to an issue that has been
> manifesting itself in the Akka streams implementation of the Kafka
> connector. When it comes to consuming messages, the implementation relies
> heavily on the fact that we can pause and resume partitions. In some
> situations when a single consumer instance is shared among several streams,
> we might end up with frequently pausing and unpausing a set of topic
> partitions, which is the main facility that allows us to implement back
> pressure. This however has certain disadvantages, especially when there are
> two consumers that differ in terms of processing speed.
>
> To articulate the issue more clearly, imagine that a consumer maintains
> assignments for two topic partitions *TP1* and *TP2*. This consumer is
> shared by two streams - S1 and S2. So effectively when we have demand from
> only one of the streams - *S1*, we will pause one of the topic partitions
> *TP2* and call *poll()* on the consumer to only retrieve the records for
> the demanded topic partition - *TP1*. The result of that is all the records
> that have been prefetched for *TP2* are now thrown away by the fetcher ("*Not
> returning fetched records for assigned partition TP2 since it is no longer
> fetchable"*). If we extrapolate that to multiple streams sharing the same
> consumer, we might quickly end up in a situation where we throw prefetched
> data quite often. This does not seem like the most efficient approach and
> in fact produces quite a lot of overlapping fetch requests as illustrated
> in the following issue:
>
> https://github.com/akka/alpakka-kafka/issues/549
>
> I am writing this email to get some initial opinion on a KIP I was thinking
> about. What if we give the clients of the Consumer API a bit more control
> of what to do with this prefetched data. Two options I am wondering about:
>
> 1. Introduce a configuration setting, such as*
> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> think of a better name), which when set to true will return what is
> prefetched instead of throwing it away on calling *poll()*. Since this is
> amount of data that is bounded by the maximum size of the prefetch, we can
> control what is the most amount of records returned. The client of the
> consumer API can then be responsible for keeping that data around and use
> it when appropriate (i.e. when demand is present)
>
> 2. Introduce a facility to pass in a buffer into which the prefetched
> records are drained when poll is called and paused partitions have some
> prefetched records.
>
> Any opinions on the matter are welcome. Thanks a lot !
>
> Zahari Dichev
>

Re: Throwing away prefetched records optimisation.

[Ryanne Dolan <ry...@gmail.com>]

Zahira,

Kafka consumers use a pull model. I'm not sure what backpressure means in
this context. If a consumer isn't ready for more records, it just doesn't
poll() for more.

The documentation talks about "flow control" but doesn't mention
"backpressure". I think these are related but different concepts.
Pause/resume lets you prioritize some topics/partitions over others ("flow
control"), but that isn't a signal to a sender to stop sending
("backpressure").

Ryanne

On Wed, Oct 17, 2018 at 1:55 PM Zahari Dichev <za...@gmail.com>
wrote:

> Hi there Ryanne,
>
> Thanks for the response ! There is most likely quite a lot that I am
> missing here, but after I read the docs, it seems to me that the
> pause/resume API has been provided with the very purpose of implementing
> bespoke flow control. That being said, I see it as quite natural to be able
> to pause and resume as needed without facing the problems outlined in
> my previous email. So if we are going totally wrong about this and using
> the pause/resume the wrong way, feel free to elaborate. I am really not
> trying to argue my case here, just genuinely attempting to understand what
> can be done on our end to improve the Akka streams integration.. Thanks in
> advance :)
>
> Zahari
>
> On Wed, Oct 17, 2018 at 5:49 PM Ryanne Dolan <ry...@gmail.com>
> wrote:
>
> > Zahari,
> >
> > It sounds to me like this problem is due to Akka attempting to implement
> > additional backpressure on top of the Consumer API. I'd suggest they not
> do
> > that, and then this problem goes away.
> >
> > Ryanne
> >
> > On Wed, Oct 17, 2018 at 7:35 AM Zahari Dichev <za...@gmail.com>
> > wrote:
> >
> > > Hi there,
> > >
> > > Are there any opinions on the matter described in my previous email? I
> > > think this is quite important when it comes to implementing any non
> > trivial
> > > functionality that relies on pause/resume. Of course if I am mistaken,
> > feel
> > > free to elaborate.
> > >
> > > Thanks,
> > > Zahari
> > >
> > > On Tue, Oct 16, 2018 at 10:29 AM Zahari Dichev <zaharidichev@gmail.com
> >
> > > wrote:
> > >
> > > > Hi there Kafka developers,
> > > >
> > > > I am currently trying to find a solution to an issue that has been
> > > > manifesting itself in the Akka streams implementation of the Kafka
> > > > connector. When it comes to consuming messages, the implementation
> > relies
> > > > heavily on the fact that we can pause and resume partitions. In some
> > > > situations when a single consumer instance is shared among several
> > > streams,
> > > > we might end up with frequently pausing and unpausing a set of topic
> > > > partitions, which is the main facility that allows us to implement
> back
> > > > pressure. This however has certain disadvantages, especially when
> there
> > > are
> > > > two consumers that differ in terms of processing speed.
> > > >
> > > > To articulate the issue more clearly, imagine that a consumer
> maintains
> > > > assignments for two topic partitions *TP1* and *TP2*. This consumer
> is
> > > > shared by two streams - S1 and S2. So effectively when we have demand
> > > from
> > > > only one of the streams - *S1*, we will pause one of the topic
> > partitions
> > > > *TP2* and call *poll()* on the consumer to only retrieve the records
> > for
> > > > the demanded topic partition - *TP1*. The result of that is all the
> > > > records that have been prefetched for *TP2* are now thrown away by
> the
> > > > fetcher ("*Not returning fetched records for assigned partition TP2
> > since
> > > > it is no longer fetchable"*). If we extrapolate that to multiple
> > streams
> > > > sharing the same consumer, we might quickly end up in a situation
> where
> > > we
> > > > throw prefetched data quite often. This does not seem like the most
> > > > efficient approach and in fact produces quite a lot of overlapping
> > fetch
> > > > requests as illustrated in the following issue:
> > > >
> > > > https://github.com/akka/alpakka-kafka/issues/549
> > > >
> > > > I am writing this email to get some initial opinion on a KIP I was
> > > > thinking about. What if we give the clients of the Consumer API a bit
> > > more
> > > > control of what to do with this prefetched data. Two options I am
> > > wondering
> > > > about:
> > > >
> > > > 1. Introduce a configuration setting, such as*
> > > > "return-prefetched-data-for-paused-topic-partitions = false"* (have
> to
> > > > think of a better name), which when set to true will return what is
> > > > prefetched instead of throwing it away on calling *poll()*. Since
> this
> > is
> > > > amount of data that is bounded by the maximum size of the prefetch,
> we
> > > can
> > > > control what is the most amount of records returned. The client of
> the
> > > > consumer API can then be responsible for keeping that data around and
> > use
> > > > it when appropriate (i.e. when demand is present)
> > > >
> > > > 2. Introduce a facility to pass in a buffer into which the prefetched
> > > > records are drained when poll is called and paused partitions have
> some
> > > > prefetched records.
> > > >
> > > > Any opinions on the matter are welcome. Thanks a lot !
> > > >
> > > > Zahari Dichev
> > > >
> > >
> >
>

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Hi there Ryanne,

Thanks for the response ! There is most likely quite a lot that I am
missing here, but after I read the docs, it seems to me that the
pause/resume API has been provided with the very purpose of implementing
bespoke flow control. That being said, I see it as quite natural to be able
to pause and resume as needed without facing the problems outlined in
my previous email. So if we are going totally wrong about this and using
the pause/resume the wrong way, feel free to elaborate. I am really not
trying to argue my case here, just genuinely attempting to understand what
can be done on our end to improve the Akka streams integration.. Thanks in
advance :)

Zahari

On Wed, Oct 17, 2018 at 5:49 PM Ryanne Dolan <ry...@gmail.com> wrote:

> Zahari,
>
> It sounds to me like this problem is due to Akka attempting to implement
> additional backpressure on top of the Consumer API. I'd suggest they not do
> that, and then this problem goes away.
>
> Ryanne
>
> On Wed, Oct 17, 2018 at 7:35 AM Zahari Dichev <za...@gmail.com>
> wrote:
>
> > Hi there,
> >
> > Are there any opinions on the matter described in my previous email? I
> > think this is quite important when it comes to implementing any non
> trivial
> > functionality that relies on pause/resume. Of course if I am mistaken,
> feel
> > free to elaborate.
> >
> > Thanks,
> > Zahari
> >
> > On Tue, Oct 16, 2018 at 10:29 AM Zahari Dichev <za...@gmail.com>
> > wrote:
> >
> > > Hi there Kafka developers,
> > >
> > > I am currently trying to find a solution to an issue that has been
> > > manifesting itself in the Akka streams implementation of the Kafka
> > > connector. When it comes to consuming messages, the implementation
> relies
> > > heavily on the fact that we can pause and resume partitions. In some
> > > situations when a single consumer instance is shared among several
> > streams,
> > > we might end up with frequently pausing and unpausing a set of topic
> > > partitions, which is the main facility that allows us to implement back
> > > pressure. This however has certain disadvantages, especially when there
> > are
> > > two consumers that differ in terms of processing speed.
> > >
> > > To articulate the issue more clearly, imagine that a consumer maintains
> > > assignments for two topic partitions *TP1* and *TP2*. This consumer is
> > > shared by two streams - S1 and S2. So effectively when we have demand
> > from
> > > only one of the streams - *S1*, we will pause one of the topic
> partitions
> > > *TP2* and call *poll()* on the consumer to only retrieve the records
> for
> > > the demanded topic partition - *TP1*. The result of that is all the
> > > records that have been prefetched for *TP2* are now thrown away by the
> > > fetcher ("*Not returning fetched records for assigned partition TP2
> since
> > > it is no longer fetchable"*). If we extrapolate that to multiple
> streams
> > > sharing the same consumer, we might quickly end up in a situation where
> > we
> > > throw prefetched data quite often. This does not seem like the most
> > > efficient approach and in fact produces quite a lot of overlapping
> fetch
> > > requests as illustrated in the following issue:
> > >
> > > https://github.com/akka/alpakka-kafka/issues/549
> > >
> > > I am writing this email to get some initial opinion on a KIP I was
> > > thinking about. What if we give the clients of the Consumer API a bit
> > more
> > > control of what to do with this prefetched data. Two options I am
> > wondering
> > > about:
> > >
> > > 1. Introduce a configuration setting, such as*
> > > "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> > > think of a better name), which when set to true will return what is
> > > prefetched instead of throwing it away on calling *poll()*. Since this
> is
> > > amount of data that is bounded by the maximum size of the prefetch, we
> > can
> > > control what is the most amount of records returned. The client of the
> > > consumer API can then be responsible for keeping that data around and
> use
> > > it when appropriate (i.e. when demand is present)
> > >
> > > 2. Introduce a facility to pass in a buffer into which the prefetched
> > > records are drained when poll is called and paused partitions have some
> > > prefetched records.
> > >
> > > Any opinions on the matter are welcome. Thanks a lot !
> > >
> > > Zahari Dichev
> > >
> >
>

Re: Throwing away prefetched records optimisation.

[Ryanne Dolan <ry...@gmail.com>]

Zahari,

It sounds to me like this problem is due to Akka attempting to implement
additional backpressure on top of the Consumer API. I'd suggest they not do
that, and then this problem goes away.

Ryanne

On Wed, Oct 17, 2018 at 7:35 AM Zahari Dichev <za...@gmail.com>
wrote:

> Hi there,
>
> Are there any opinions on the matter described in my previous email? I
> think this is quite important when it comes to implementing any non trivial
> functionality that relies on pause/resume. Of course if I am mistaken, feel
> free to elaborate.
>
> Thanks,
> Zahari
>
> On Tue, Oct 16, 2018 at 10:29 AM Zahari Dichev <za...@gmail.com>
> wrote:
>
> > Hi there Kafka developers,
> >
> > I am currently trying to find a solution to an issue that has been
> > manifesting itself in the Akka streams implementation of the Kafka
> > connector. When it comes to consuming messages, the implementation relies
> > heavily on the fact that we can pause and resume partitions. In some
> > situations when a single consumer instance is shared among several
> streams,
> > we might end up with frequently pausing and unpausing a set of topic
> > partitions, which is the main facility that allows us to implement back
> > pressure. This however has certain disadvantages, especially when there
> are
> > two consumers that differ in terms of processing speed.
> >
> > To articulate the issue more clearly, imagine that a consumer maintains
> > assignments for two topic partitions *TP1* and *TP2*. This consumer is
> > shared by two streams - S1 and S2. So effectively when we have demand
> from
> > only one of the streams - *S1*, we will pause one of the topic partitions
> > *TP2* and call *poll()* on the consumer to only retrieve the records for
> > the demanded topic partition - *TP1*. The result of that is all the
> > records that have been prefetched for *TP2* are now thrown away by the
> > fetcher ("*Not returning fetched records for assigned partition TP2 since
> > it is no longer fetchable"*). If we extrapolate that to multiple streams
> > sharing the same consumer, we might quickly end up in a situation where
> we
> > throw prefetched data quite often. This does not seem like the most
> > efficient approach and in fact produces quite a lot of overlapping fetch
> > requests as illustrated in the following issue:
> >
> > https://github.com/akka/alpakka-kafka/issues/549
> >
> > I am writing this email to get some initial opinion on a KIP I was
> > thinking about. What if we give the clients of the Consumer API a bit
> more
> > control of what to do with this prefetched data. Two options I am
> wondering
> > about:
> >
> > 1. Introduce a configuration setting, such as*
> > "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> > think of a better name), which when set to true will return what is
> > prefetched instead of throwing it away on calling *poll()*. Since this is
> > amount of data that is bounded by the maximum size of the prefetch, we
> can
> > control what is the most amount of records returned. The client of the
> > consumer API can then be responsible for keeping that data around and use
> > it when appropriate (i.e. when demand is present)
> >
> > 2. Introduce a facility to pass in a buffer into which the prefetched
> > records are drained when poll is called and paused partitions have some
> > prefetched records.
> >
> > Any opinions on the matter are welcome. Thanks a lot !
> >
> > Zahari Dichev
> >
>

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Hi there,

Are there any opinions on the matter described in my previous email? I
think this is quite important when it comes to implementing any non trivial
functionality that relies on pause/resume. Of course if I am mistaken, feel
free to elaborate.

Thanks,
Zahari

On Tue, Oct 16, 2018 at 10:29 AM Zahari Dichev <za...@gmail.com>
wrote:

> Hi there Kafka developers,
>
> I am currently trying to find a solution to an issue that has been
> manifesting itself in the Akka streams implementation of the Kafka
> connector. When it comes to consuming messages, the implementation relies
> heavily on the fact that we can pause and resume partitions. In some
> situations when a single consumer instance is shared among several streams,
> we might end up with frequently pausing and unpausing a set of topic
> partitions, which is the main facility that allows us to implement back
> pressure. This however has certain disadvantages, especially when there are
> two consumers that differ in terms of processing speed.
>
> To articulate the issue more clearly, imagine that a consumer maintains
> assignments for two topic partitions *TP1* and *TP2*. This consumer is
> shared by two streams - S1 and S2. So effectively when we have demand from
> only one of the streams - *S1*, we will pause one of the topic partitions
> *TP2* and call *poll()* on the consumer to only retrieve the records for
> the demanded topic partition - *TP1*. The result of that is all the
> records that have been prefetched for *TP2* are now thrown away by the
> fetcher ("*Not returning fetched records for assigned partition TP2 since
> it is no longer fetchable"*). If we extrapolate that to multiple streams
> sharing the same consumer, we might quickly end up in a situation where we
> throw prefetched data quite often. This does not seem like the most
> efficient approach and in fact produces quite a lot of overlapping fetch
> requests as illustrated in the following issue:
>
> https://github.com/akka/alpakka-kafka/issues/549
>
> I am writing this email to get some initial opinion on a KIP I was
> thinking about. What if we give the clients of the Consumer API a bit more
> control of what to do with this prefetched data. Two options I am wondering
> about:
>
> 1. Introduce a configuration setting, such as*
> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> think of a better name), which when set to true will return what is
> prefetched instead of throwing it away on calling *poll()*. Since this is
> amount of data that is bounded by the maximum size of the prefetch, we can
> control what is the most amount of records returned. The client of the
> consumer API can then be responsible for keeping that data around and use
> it when appropriate (i.e. when demand is present)
>
> 2. Introduce a facility to pass in a buffer into which the prefetched
> records are drained when poll is called and paused partitions have some
> prefetched records.
>
> Any opinions on the matter are welcome. Thanks a lot !
>
> Zahari Dichev
>

Re: Throwing away prefetched records optimisation.

[Colin McCabe <cm...@fastmail.com>]

In general, the official discussion thread for a KIP starts after the KIP is written and posted.  So you would typically start a new email thread with a [DISCUSS] string in the title somewhere.  You can certainly link back to this email thread if you want, though, since it has some useful context for everything.

best,
Colin


On Tue, Oct 23, 2018, at 23:21, Zahari Dichev wrote:
> Colin, I agree
> 
> I will take a closer looks at the Fetcher itself and see whether that is
> feasible and update my KIP accordingly. I guess we can label this one, the
> official discussion thread for it or should I start another one ?
> 
> Zahari
> 
> On Wed, Oct 24, 2018 at 6:43 AM Colin McCabe <cm...@apache.org> wrote:
> 
> > On Tue, Oct 23, 2018, at 12:38, Zahari Dichev wrote:
> > > Hi there Matthias, I looked through the code of Kafka Streams. Quite
> > > impressive work ! If I have to put the logic of buffering within the
> > > context of what we are doing in Akka though, I might end up with the
> > > following situation.
> > >
> > > 1. Poll is called with two partition being active *TP1, TP2*
> > > 2. We get some data for both, both of them also prefetch some data.
> > > 3. So now we have some data that we have obtained and some data that sits
> > > with the buffer of the fetcher, waiting to be obtained.
> > > 4. We put the data that we have obtained from the poll into the
> > respective
> > > buffers of the partitions.
> > > 5. Since both of our buffers are "full", we call pause on both *TP1* and
> > > *TP2*.
> > > 6. A little time has passed and the client of *TP1* has processed all its
> > > records from the buffer, while the one of *TP2* has processed none
> > > 7. Buffer of *TP1* gets empty, we call resume on *TP1*
> > > 8. We call poll again with *TP1* resumed and *TP2* paused.
> > > 9. We get some records for TP1 and we throw away all the records that
> > were
> > > prefetched for *TP2* in step 2
> > >
> > > This can go on and on and due to the dynamic nature of the speed of
> > > processing records and the theoretically unlimited number of topic
> > > partitions, I find it possible that this scenario can happen more than
> > once
> > > over the lifetime of a client. And instead of trying to calculate the
> > > probability of this happening and attempt to minimise it, I would prefer
> > to
> > > have one of two options:
> > >
> > > 1. Having control to allow me to enable the returning of already
> > prefetched
> > > data, and simply store it in a buffer of my own until I have enough
> > > capacity to deal with it
> > >
> > > OR
> > >
> > > 2. Keep the data in the fetcher and not throw it away but use it on the
> > > next poll (not sure how viable that is as I have not looked at the
> > details
> > > of it all).
> >
> > I haven't thought about it that hard, but it sounds like the second option
> > might be better.  I have a hard time thinking of a case where we actually
> > want to throw away data for paused partitions.  If you're still subscribed
> > to it, presumably you'll eventually unpause it and use the cache, right?
> > It makes sense for unsubscribe to clear those records, but not pause, as
> > far as I can see.
> >
> > best,
> > Colin
> >
> >
> > >
> > > The first option is what I suggested initially and the second option is
> > the
> > > one that will allow us to skip the introduction of a configuration
> > > parameter as Colin suggested. These are the things I can suggest at the
> > > moment. As mentioned, I am willing to carry out the work. There is also
> > an
> > > official discussion thread, but I guess we have deviated from that, so I
> > > can just put that current on in JIRA instead if that is OK ?
> > >
> > > Matthias, regarding how the fetcher works. From what I have looked at,
> > > whenever the consumer polls and returns some data, we immediately issue
> > > another fetch request that delivered us records that are returned on the
> > > next poll. All these fetched records, that have not made it to the caller
> > > of poll but have been fetched are thrown away in case at the time of the
> > > nest poll() the partition is in paused state. This is what is causing the
> > > inefficiency.
> > >
> > > Any more comments are welcome.
> > >
> > > On Mon, Oct 22, 2018 at 6:00 AM Ismael Juma <is...@gmail.com> wrote:
> > >
> > > > Hi,
> > > >
> > > > I think a KIP to discuss a concrete proposal makes sense. One
> > suggestion is
> > > > to explore the possibility of fixing the issue without a new config.
> > Would
> > > > that break existing users? Generally, we should strive for avoiding
> > configs
> > > > if at all possible.
> > > >
> > > > Ismael
> > > >
> > > > On 16 Oct 2018 12:30 am, "Zahari Dichev" <za...@gmail.com>
> > wrote:
> > > >
> > > > Hi there Kafka developers,
> > > >
> > > > I am currently trying to find a solution to an issue that has been
> > > > manifesting itself in the Akka streams implementation of the Kafka
> > > > connector. When it comes to consuming messages, the implementation
> > relies
> > > > heavily on the fact that we can pause and resume partitions. In some
> > > > situations when a single consumer instance is shared among several
> > streams,
> > > > we might end up with frequently pausing and unpausing a set of topic
> > > > partitions, which is the main facility that allows us to implement back
> > > > pressure. This however has certain disadvantages, especially when
> > there are
> > > > two consumers that differ in terms of processing speed.
> > > >
> > > > To articulate the issue more clearly, imagine that a consumer maintains
> > > > assignments for two topic partitions *TP1* and *TP2*. This consumer is
> > > > shared by two streams - S1 and S2. So effectively when we have demand
> > from
> > > > only one of the streams - *S1*, we will pause one of the topic
> > partitions
> > > > *TP2* and call *poll()* on the consumer to only retrieve the records
> > for
> > > > the demanded topic partition - *TP1*. The result of that is all the
> > records
> > > > that have been prefetched for *TP2* are now thrown away by the fetcher
> > > > ("*Not
> > > > returning fetched records for assigned partition TP2 since it is no
> > longer
> > > > fetchable"*). If we extrapolate that to multiple streams sharing the
> > same
> > > > consumer, we might quickly end up in a situation where we throw
> > prefetched
> > > > data quite often. This does not seem like the most efficient approach
> > and
> > > > in fact produces quite a lot of overlapping fetch requests as
> > illustrated
> > > > in the following issue:
> > > >
> > > > https://github.com/akka/alpakka-kafka/issues/549
> > > >
> > > > I am writing this email to get some initial opinion on a KIP I was
> > thinking
> > > > about. What if we give the clients of the Consumer API a bit more
> > control
> > > > of what to do with this prefetched data. Two options I am wondering
> > about:
> > > >
> > > > 1. Introduce a configuration setting, such as*
> > > > "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> > > > think of a better name), which when set to true will return what is
> > > > prefetched instead of throwing it away on calling *poll()*. Since this
> > is
> > > > amount of data that is bounded by the maximum size of the prefetch, we
> > can
> > > > control what is the most amount of records returned. The client of the
> > > > consumer API can then be responsible for keeping that data around and
> > use
> > > > it when appropriate (i.e. when demand is present)
> > > >
> > > > 2. Introduce a facility to pass in a buffer into which the prefetched
> > > > records are drained when poll is called and paused partitions have some
> > > > prefetched records.
> > > >
> > > > Any opinions on the matter are welcome. Thanks a lot !
> > > >
> > > >
> > > > Zahari Dichev
> > > >
> >

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Colin, I agree

I will take a closer looks at the Fetcher itself and see whether that is
feasible and update my KIP accordingly. I guess we can label this one, the
official discussion thread for it or should I start another one ?

Zahari

On Wed, Oct 24, 2018 at 6:43 AM Colin McCabe <cm...@apache.org> wrote:

> On Tue, Oct 23, 2018, at 12:38, Zahari Dichev wrote:
> > Hi there Matthias, I looked through the code of Kafka Streams. Quite
> > impressive work ! If I have to put the logic of buffering within the
> > context of what we are doing in Akka though, I might end up with the
> > following situation.
> >
> > 1. Poll is called with two partition being active *TP1, TP2*
> > 2. We get some data for both, both of them also prefetch some data.
> > 3. So now we have some data that we have obtained and some data that sits
> > with the buffer of the fetcher, waiting to be obtained.
> > 4. We put the data that we have obtained from the poll into the
> respective
> > buffers of the partitions.
> > 5. Since both of our buffers are "full", we call pause on both *TP1* and
> > *TP2*.
> > 6. A little time has passed and the client of *TP1* has processed all its
> > records from the buffer, while the one of *TP2* has processed none
> > 7. Buffer of *TP1* gets empty, we call resume on *TP1*
> > 8. We call poll again with *TP1* resumed and *TP2* paused.
> > 9. We get some records for TP1 and we throw away all the records that
> were
> > prefetched for *TP2* in step 2
> >
> > This can go on and on and due to the dynamic nature of the speed of
> > processing records and the theoretically unlimited number of topic
> > partitions, I find it possible that this scenario can happen more than
> once
> > over the lifetime of a client. And instead of trying to calculate the
> > probability of this happening and attempt to minimise it, I would prefer
> to
> > have one of two options:
> >
> > 1. Having control to allow me to enable the returning of already
> prefetched
> > data, and simply store it in a buffer of my own until I have enough
> > capacity to deal with it
> >
> > OR
> >
> > 2. Keep the data in the fetcher and not throw it away but use it on the
> > next poll (not sure how viable that is as I have not looked at the
> details
> > of it all).
>
> I haven't thought about it that hard, but it sounds like the second option
> might be better.  I have a hard time thinking of a case where we actually
> want to throw away data for paused partitions.  If you're still subscribed
> to it, presumably you'll eventually unpause it and use the cache, right?
> It makes sense for unsubscribe to clear those records, but not pause, as
> far as I can see.
>
> best,
> Colin
>
>
> >
> > The first option is what I suggested initially and the second option is
> the
> > one that will allow us to skip the introduction of a configuration
> > parameter as Colin suggested. These are the things I can suggest at the
> > moment. As mentioned, I am willing to carry out the work. There is also
> an
> > official discussion thread, but I guess we have deviated from that, so I
> > can just put that current on in JIRA instead if that is OK ?
> >
> > Matthias, regarding how the fetcher works. From what I have looked at,
> > whenever the consumer polls and returns some data, we immediately issue
> > another fetch request that delivered us records that are returned on the
> > next poll. All these fetched records, that have not made it to the caller
> > of poll but have been fetched are thrown away in case at the time of the
> > nest poll() the partition is in paused state. This is what is causing the
> > inefficiency.
> >
> > Any more comments are welcome.
> >
> > On Mon, Oct 22, 2018 at 6:00 AM Ismael Juma <is...@gmail.com> wrote:
> >
> > > Hi,
> > >
> > > I think a KIP to discuss a concrete proposal makes sense. One
> suggestion is
> > > to explore the possibility of fixing the issue without a new config.
> Would
> > > that break existing users? Generally, we should strive for avoiding
> configs
> > > if at all possible.
> > >
> > > Ismael
> > >
> > > On 16 Oct 2018 12:30 am, "Zahari Dichev" <za...@gmail.com>
> wrote:
> > >
> > > Hi there Kafka developers,
> > >
> > > I am currently trying to find a solution to an issue that has been
> > > manifesting itself in the Akka streams implementation of the Kafka
> > > connector. When it comes to consuming messages, the implementation
> relies
> > > heavily on the fact that we can pause and resume partitions. In some
> > > situations when a single consumer instance is shared among several
> streams,
> > > we might end up with frequently pausing and unpausing a set of topic
> > > partitions, which is the main facility that allows us to implement back
> > > pressure. This however has certain disadvantages, especially when
> there are
> > > two consumers that differ in terms of processing speed.
> > >
> > > To articulate the issue more clearly, imagine that a consumer maintains
> > > assignments for two topic partitions *TP1* and *TP2*. This consumer is
> > > shared by two streams - S1 and S2. So effectively when we have demand
> from
> > > only one of the streams - *S1*, we will pause one of the topic
> partitions
> > > *TP2* and call *poll()* on the consumer to only retrieve the records
> for
> > > the demanded topic partition - *TP1*. The result of that is all the
> records
> > > that have been prefetched for *TP2* are now thrown away by the fetcher
> > > ("*Not
> > > returning fetched records for assigned partition TP2 since it is no
> longer
> > > fetchable"*). If we extrapolate that to multiple streams sharing the
> same
> > > consumer, we might quickly end up in a situation where we throw
> prefetched
> > > data quite often. This does not seem like the most efficient approach
> and
> > > in fact produces quite a lot of overlapping fetch requests as
> illustrated
> > > in the following issue:
> > >
> > > https://github.com/akka/alpakka-kafka/issues/549
> > >
> > > I am writing this email to get some initial opinion on a KIP I was
> thinking
> > > about. What if we give the clients of the Consumer API a bit more
> control
> > > of what to do with this prefetched data. Two options I am wondering
> about:
> > >
> > > 1. Introduce a configuration setting, such as*
> > > "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> > > think of a better name), which when set to true will return what is
> > > prefetched instead of throwing it away on calling *poll()*. Since this
> is
> > > amount of data that is bounded by the maximum size of the prefetch, we
> can
> > > control what is the most amount of records returned. The client of the
> > > consumer API can then be responsible for keeping that data around and
> use
> > > it when appropriate (i.e. when demand is present)
> > >
> > > 2. Introduce a facility to pass in a buffer into which the prefetched
> > > records are drained when poll is called and paused partitions have some
> > > prefetched records.
> > >
> > > Any opinions on the matter are welcome. Thanks a lot !
> > >
> > >
> > > Zahari Dichev
> > >
>

Re: Throwing away prefetched records optimisation.

[Colin McCabe <cm...@apache.org>]

On Tue, Oct 23, 2018, at 12:38, Zahari Dichev wrote:
> Hi there Matthias, I looked through the code of Kafka Streams. Quite
> impressive work ! If I have to put the logic of buffering within the
> context of what we are doing in Akka though, I might end up with the
> following situation.
> 
> 1. Poll is called with two partition being active *TP1, TP2*
> 2. We get some data for both, both of them also prefetch some data.
> 3. So now we have some data that we have obtained and some data that sits
> with the buffer of the fetcher, waiting to be obtained.
> 4. We put the data that we have obtained from the poll into the respective
> buffers of the partitions.
> 5. Since both of our buffers are "full", we call pause on both *TP1* and
> *TP2*.
> 6. A little time has passed and the client of *TP1* has processed all its
> records from the buffer, while the one of *TP2* has processed none
> 7. Buffer of *TP1* gets empty, we call resume on *TP1*
> 8. We call poll again with *TP1* resumed and *TP2* paused.
> 9. We get some records for TP1 and we throw away all the records that were
> prefetched for *TP2* in step 2
> 
> This can go on and on and due to the dynamic nature of the speed of
> processing records and the theoretically unlimited number of topic
> partitions, I find it possible that this scenario can happen more than once
> over the lifetime of a client. And instead of trying to calculate the
> probability of this happening and attempt to minimise it, I would prefer to
> have one of two options:
> 
> 1. Having control to allow me to enable the returning of already prefetched
> data, and simply store it in a buffer of my own until I have enough
> capacity to deal with it
> 
> OR
> 
> 2. Keep the data in the fetcher and not throw it away but use it on the
> next poll (not sure how viable that is as I have not looked at the details
> of it all).

I haven't thought about it that hard, but it sounds like the second option might be better.  I have a hard time thinking of a case where we actually want to throw away data for paused partitions.  If you're still subscribed to it, presumably you'll eventually unpause it and use the cache, right?  It makes sense for unsubscribe to clear those records, but not pause, as far as I can see.

best,
Colin


> 
> The first option is what I suggested initially and the second option is the
> one that will allow us to skip the introduction of a configuration
> parameter as Colin suggested. These are the things I can suggest at the
> moment. As mentioned, I am willing to carry out the work. There is also an
> official discussion thread, but I guess we have deviated from that, so I
> can just put that current on in JIRA instead if that is OK ?
> 
> Matthias, regarding how the fetcher works. From what I have looked at,
> whenever the consumer polls and returns some data, we immediately issue
> another fetch request that delivered us records that are returned on the
> next poll. All these fetched records, that have not made it to the caller
> of poll but have been fetched are thrown away in case at the time of the
> nest poll() the partition is in paused state. This is what is causing the
> inefficiency.
> 
> Any more comments are welcome.
> 
> On Mon, Oct 22, 2018 at 6:00 AM Ismael Juma <is...@gmail.com> wrote:
> 
> > Hi,
> >
> > I think a KIP to discuss a concrete proposal makes sense. One suggestion is
> > to explore the possibility of fixing the issue without a new config. Would
> > that break existing users? Generally, we should strive for avoiding configs
> > if at all possible.
> >
> > Ismael
> >
> > On 16 Oct 2018 12:30 am, "Zahari Dichev" <za...@gmail.com> wrote:
> >
> > Hi there Kafka developers,
> >
> > I am currently trying to find a solution to an issue that has been
> > manifesting itself in the Akka streams implementation of the Kafka
> > connector. When it comes to consuming messages, the implementation relies
> > heavily on the fact that we can pause and resume partitions. In some
> > situations when a single consumer instance is shared among several streams,
> > we might end up with frequently pausing and unpausing a set of topic
> > partitions, which is the main facility that allows us to implement back
> > pressure. This however has certain disadvantages, especially when there are
> > two consumers that differ in terms of processing speed.
> >
> > To articulate the issue more clearly, imagine that a consumer maintains
> > assignments for two topic partitions *TP1* and *TP2*. This consumer is
> > shared by two streams - S1 and S2. So effectively when we have demand from
> > only one of the streams - *S1*, we will pause one of the topic partitions
> > *TP2* and call *poll()* on the consumer to only retrieve the records for
> > the demanded topic partition - *TP1*. The result of that is all the records
> > that have been prefetched for *TP2* are now thrown away by the fetcher
> > ("*Not
> > returning fetched records for assigned partition TP2 since it is no longer
> > fetchable"*). If we extrapolate that to multiple streams sharing the same
> > consumer, we might quickly end up in a situation where we throw prefetched
> > data quite often. This does not seem like the most efficient approach and
> > in fact produces quite a lot of overlapping fetch requests as illustrated
> > in the following issue:
> >
> > https://github.com/akka/alpakka-kafka/issues/549
> >
> > I am writing this email to get some initial opinion on a KIP I was thinking
> > about. What if we give the clients of the Consumer API a bit more control
> > of what to do with this prefetched data. Two options I am wondering about:
> >
> > 1. Introduce a configuration setting, such as*
> > "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> > think of a better name), which when set to true will return what is
> > prefetched instead of throwing it away on calling *poll()*. Since this is
> > amount of data that is bounded by the maximum size of the prefetch, we can
> > control what is the most amount of records returned. The client of the
> > consumer API can then be responsible for keeping that data around and use
> > it when appropriate (i.e. when demand is present)
> >
> > 2. Introduce a facility to pass in a buffer into which the prefetched
> > records are drained when poll is called and paused partitions have some
> > prefetched records.
> >
> > Any opinions on the matter are welcome. Thanks a lot !
> >
> >
> > Zahari Dichev
> >

Re: Throwing away prefetched records optimisation.

["Matthias J. Sax" <ma...@confluent.io>]

Thanks for the detailed explanation.

-Matthias

On 10/23/18 12:38 PM, Zahari Dichev wrote:
> Hi there Matthias, I looked through the code of Kafka Streams. Quite
> impressive work ! If I have to put the logic of buffering within the
> context of what we are doing in Akka though, I might end up with the
> following situation.
> 
> 1. Poll is called with two partition being active *TP1, TP2*
> 2. We get some data for both, both of them also prefetch some data.
> 3. So now we have some data that we have obtained and some data that sits
> with the buffer of the fetcher, waiting to be obtained.
> 4. We put the data that we have obtained from the poll into the respective
> buffers of the partitions.
> 5. Since both of our buffers are "full", we call pause on both *TP1* and
> *TP2*.
> 6. A little time has passed and the client of *TP1* has processed all its
> records from the buffer, while the one of *TP2* has processed none
> 7. Buffer of *TP1* gets empty, we call resume on *TP1*
> 8. We call poll again with *TP1* resumed and *TP2* paused.
> 9. We get some records for TP1 and we throw away all the records that were
> prefetched for *TP2* in step 2
> 
> This can go on and on and due to the dynamic nature of the speed of
> processing records and the theoretically unlimited number of topic
> partitions, I find it possible that this scenario can happen more than once
> over the lifetime of a client. And instead of trying to calculate the
> probability of this happening and attempt to minimise it, I would prefer to
> have one of two options:
> 
> 1. Having control to allow me to enable the returning of already prefetched
> data, and simply store it in a buffer of my own until I have enough
> capacity to deal with it
> 
> OR
> 
> 2. Keep the data in the fetcher and not throw it away but use it on the
> next poll (not sure how viable that is as I have not looked at the details
> of it all).
> 
> The first option is what I suggested initially and the second option is the
> one that will allow us to skip the introduction of a configuration
> parameter as Colin suggested. These are the things I can suggest at the
> moment. As mentioned, I am willing to carry out the work. There is also an
> official discussion thread, but I guess we have deviated from that, so I
> can just put that current on in JIRA instead if that is OK ?
> 
> Matthias, regarding how the fetcher works. From what I have looked at,
> whenever the consumer polls and returns some data, we immediately issue
> another fetch request that delivered us records that are returned on the
> next poll. All these fetched records, that have not made it to the caller
> of poll but have been fetched are thrown away in case at the time of the
> nest poll() the partition is in paused state. This is what is causing the
> inefficiency.
> 
> Any more comments are welcome.
> 
> On Mon, Oct 22, 2018 at 6:00 AM Ismael Juma <is...@gmail.com> wrote:
> 
>> Hi,
>>
>> I think a KIP to discuss a concrete proposal makes sense. One suggestion is
>> to explore the possibility of fixing the issue without a new config. Would
>> that break existing users? Generally, we should strive for avoiding configs
>> if at all possible.
>>
>> Ismael
>>
>> On 16 Oct 2018 12:30 am, "Zahari Dichev" <za...@gmail.com> wrote:
>>
>> Hi there Kafka developers,
>>
>> I am currently trying to find a solution to an issue that has been
>> manifesting itself in the Akka streams implementation of the Kafka
>> connector. When it comes to consuming messages, the implementation relies
>> heavily on the fact that we can pause and resume partitions. In some
>> situations when a single consumer instance is shared among several streams,
>> we might end up with frequently pausing and unpausing a set of topic
>> partitions, which is the main facility that allows us to implement back
>> pressure. This however has certain disadvantages, especially when there are
>> two consumers that differ in terms of processing speed.
>>
>> To articulate the issue more clearly, imagine that a consumer maintains
>> assignments for two topic partitions *TP1* and *TP2*. This consumer is
>> shared by two streams - S1 and S2. So effectively when we have demand from
>> only one of the streams - *S1*, we will pause one of the topic partitions
>> *TP2* and call *poll()* on the consumer to only retrieve the records for
>> the demanded topic partition - *TP1*. The result of that is all the records
>> that have been prefetched for *TP2* are now thrown away by the fetcher
>> ("*Not
>> returning fetched records for assigned partition TP2 since it is no longer
>> fetchable"*). If we extrapolate that to multiple streams sharing the same
>> consumer, we might quickly end up in a situation where we throw prefetched
>> data quite often. This does not seem like the most efficient approach and
>> in fact produces quite a lot of overlapping fetch requests as illustrated
>> in the following issue:
>>
>> https://github.com/akka/alpakka-kafka/issues/549
>>
>> I am writing this email to get some initial opinion on a KIP I was thinking
>> about. What if we give the clients of the Consumer API a bit more control
>> of what to do with this prefetched data. Two options I am wondering about:
>>
>> 1. Introduce a configuration setting, such as*
>> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
>> think of a better name), which when set to true will return what is
>> prefetched instead of throwing it away on calling *poll()*. Since this is
>> amount of data that is bounded by the maximum size of the prefetch, we can
>> control what is the most amount of records returned. The client of the
>> consumer API can then be responsible for keeping that data around and use
>> it when appropriate (i.e. when demand is present)
>>
>> 2. Introduce a facility to pass in a buffer into which the prefetched
>> records are drained when poll is called and paused partitions have some
>> prefetched records.
>>
>> Any opinions on the matter are welcome. Thanks a lot !
>>
>>
>> Zahari Dichev
>>
> 

Re: Throwing away prefetched records optimisation.

[Zahari Dichev <za...@gmail.com>]

Hi there Matthias, I looked through the code of Kafka Streams. Quite
impressive work ! If I have to put the logic of buffering within the
context of what we are doing in Akka though, I might end up with the
following situation.

1. Poll is called with two partition being active *TP1, TP2*
2. We get some data for both, both of them also prefetch some data.
3. So now we have some data that we have obtained and some data that sits
with the buffer of the fetcher, waiting to be obtained.
4. We put the data that we have obtained from the poll into the respective
buffers of the partitions.
5. Since both of our buffers are "full", we call pause on both *TP1* and
*TP2*.
6. A little time has passed and the client of *TP1* has processed all its
records from the buffer, while the one of *TP2* has processed none
7. Buffer of *TP1* gets empty, we call resume on *TP1*
8. We call poll again with *TP1* resumed and *TP2* paused.
9. We get some records for TP1 and we throw away all the records that were
prefetched for *TP2* in step 2

This can go on and on and due to the dynamic nature of the speed of
processing records and the theoretically unlimited number of topic
partitions, I find it possible that this scenario can happen more than once
over the lifetime of a client. And instead of trying to calculate the
probability of this happening and attempt to minimise it, I would prefer to
have one of two options:

1. Having control to allow me to enable the returning of already prefetched
data, and simply store it in a buffer of my own until I have enough
capacity to deal with it

OR

2. Keep the data in the fetcher and not throw it away but use it on the
next poll (not sure how viable that is as I have not looked at the details
of it all).

The first option is what I suggested initially and the second option is the
one that will allow us to skip the introduction of a configuration
parameter as Colin suggested. These are the things I can suggest at the
moment. As mentioned, I am willing to carry out the work. There is also an
official discussion thread, but I guess we have deviated from that, so I
can just put that current on in JIRA instead if that is OK ?

Matthias, regarding how the fetcher works. From what I have looked at,
whenever the consumer polls and returns some data, we immediately issue
another fetch request that delivered us records that are returned on the
next poll. All these fetched records, that have not made it to the caller
of poll but have been fetched are thrown away in case at the time of the
nest poll() the partition is in paused state. This is what is causing the
inefficiency.

Any more comments are welcome.

On Mon, Oct 22, 2018 at 6:00 AM Ismael Juma <is...@gmail.com> wrote:

> Hi,
>
> I think a KIP to discuss a concrete proposal makes sense. One suggestion is
> to explore the possibility of fixing the issue without a new config. Would
> that break existing users? Generally, we should strive for avoiding configs
> if at all possible.
>
> Ismael
>
> On 16 Oct 2018 12:30 am, "Zahari Dichev" <za...@gmail.com> wrote:
>
> Hi there Kafka developers,
>
> I am currently trying to find a solution to an issue that has been
> manifesting itself in the Akka streams implementation of the Kafka
> connector. When it comes to consuming messages, the implementation relies
> heavily on the fact that we can pause and resume partitions. In some
> situations when a single consumer instance is shared among several streams,
> we might end up with frequently pausing and unpausing a set of topic
> partitions, which is the main facility that allows us to implement back
> pressure. This however has certain disadvantages, especially when there are
> two consumers that differ in terms of processing speed.
>
> To articulate the issue more clearly, imagine that a consumer maintains
> assignments for two topic partitions *TP1* and *TP2*. This consumer is
> shared by two streams - S1 and S2. So effectively when we have demand from
> only one of the streams - *S1*, we will pause one of the topic partitions
> *TP2* and call *poll()* on the consumer to only retrieve the records for
> the demanded topic partition - *TP1*. The result of that is all the records
> that have been prefetched for *TP2* are now thrown away by the fetcher
> ("*Not
> returning fetched records for assigned partition TP2 since it is no longer
> fetchable"*). If we extrapolate that to multiple streams sharing the same
> consumer, we might quickly end up in a situation where we throw prefetched
> data quite often. This does not seem like the most efficient approach and
> in fact produces quite a lot of overlapping fetch requests as illustrated
> in the following issue:
>
> https://github.com/akka/alpakka-kafka/issues/549
>
> I am writing this email to get some initial opinion on a KIP I was thinking
> about. What if we give the clients of the Consumer API a bit more control
> of what to do with this prefetched data. Two options I am wondering about:
>
> 1. Introduce a configuration setting, such as*
> "return-prefetched-data-for-paused-topic-partitions = false"* (have to
> think of a better name), which when set to true will return what is
> prefetched instead of throwing it away on calling *poll()*. Since this is
> amount of data that is bounded by the maximum size of the prefetch, we can
> control what is the most amount of records returned. The client of the
> consumer API can then be responsible for keeping that data around and use
> it when appropriate (i.e. when demand is present)
>
> 2. Introduce a facility to pass in a buffer into which the prefetched
> records are drained when poll is called and paused partitions have some
> prefetched records.
>
> Any opinions on the matter are welcome. Thanks a lot !
>
>
> Zahari Dichev
>

Re: Throwing away prefetched records optimisation.

[Ismael Juma <is...@gmail.com>]

Hi,

I think a KIP to discuss a concrete proposal makes sense. One suggestion is
to explore the possibility of fixing the issue without a new config. Would
that break existing users? Generally, we should strive for avoiding configs
if at all possible.

Ismael

On 16 Oct 2018 12:30 am, "Zahari Dichev" <za...@gmail.com> wrote:

Hi there Kafka developers,

I am currently trying to find a solution to an issue that has been
manifesting itself in the Akka streams implementation of the Kafka
connector. When it comes to consuming messages, the implementation relies
heavily on the fact that we can pause and resume partitions. In some
situations when a single consumer instance is shared among several streams,
we might end up with frequently pausing and unpausing a set of topic
partitions, which is the main facility that allows us to implement back
pressure. This however has certain disadvantages, especially when there are
two consumers that differ in terms of processing speed.

To articulate the issue more clearly, imagine that a consumer maintains
assignments for two topic partitions *TP1* and *TP2*. This consumer is
shared by two streams - S1 and S2. So effectively when we have demand from
only one of the streams - *S1*, we will pause one of the topic partitions
*TP2* and call *poll()* on the consumer to only retrieve the records for
the demanded topic partition - *TP1*. The result of that is all the records
that have been prefetched for *TP2* are now thrown away by the fetcher
("*Not
returning fetched records for assigned partition TP2 since it is no longer
fetchable"*). If we extrapolate that to multiple streams sharing the same
consumer, we might quickly end up in a situation where we throw prefetched
data quite often. This does not seem like the most efficient approach and
in fact produces quite a lot of overlapping fetch requests as illustrated
in the following issue:

https://github.com/akka/alpakka-kafka/issues/549

I am writing this email to get some initial opinion on a KIP I was thinking
about. What if we give the clients of the Consumer API a bit more control
of what to do with this prefetched data. Two options I am wondering about:

1. Introduce a configuration setting, such as*
"return-prefetched-data-for-paused-topic-partitions = false"* (have to
think of a better name), which when set to true will return what is
prefetched instead of throwing it away on calling *poll()*. Since this is
amount of data that is bounded by the maximum size of the prefetch, we can
control what is the most amount of records returned. The client of the
consumer API can then be responsible for keeping that data around and use
it when appropriate (i.e. when demand is present)

2. Introduce a facility to pass in a buffer into which the prefetched
records are drained when poll is called and paused partitions have some
prefetched records.

Any opinions on the matter are welcome. Thanks a lot !


Zahari Dichev