Hi Bogdan, It sounds as if you could implement a form of signaling between the consumers using a distributed barrier. This can be implemented using Kafka topics. For example you could create a control thread that posts the current high-water mark for all consumers into a special topic, which gives consumers "permission" to advance to that point. Let's say your high-water mark is base on message timestamps. Here's how the consumers behave:
1.) Read the high-water mark from the signaling topic. 2.) Consume all messages up to the timestamp from the high-water mark. 3.) Go back to #1. To make this work the consumers need to let the control thread know where they are so it can post new high-water marks. Consumers therefore need to post their latest position periodically (perhaps into a second topic) so that the control thread can compute the current hindmost position and advance the barrier whenever the consumers are within some delta of each other, say 60 seconds or whatever you decide is appropriate. This algorithm has a couple of tricky elements. 1.) Stalls. If either a consumer or the control thread stops everybody does because the barrier will stop moving. Correctness is therefore key. I would monitor for failure to advance as you will undoubtedly encounter it. 2.) Problems computing the consumer position. The nastiest one is if your timestamps walk backwards as might be the case when changing from summer to winter time, when there is an NTP failure on a producing host, or if producing code has bugs. Sadly this happens more often than you would think. Cheers, Robert On Mon, Jun 9, 2014 at 4:07 AM, Bogdan Dimitriu (bdimitri) < bdimi...@cisco.com> wrote: > Certainly. > I know this may not sound like a great idea but I am running out of > options here: I¹m basically trying to implement a consumer throttle. My > application consumes from a fairly high number of partitions from a number > of consumer servers. The data is put in the partitions by a producer in a > round robin fashion so the number of messages each partition is given is > even. The messages have a time component assigned to them. > Now, for a good majority of time the consumers will be faster than the > producers, so the lags I get with the ConsumerOffsetChecker are mostly 0 > (or 1) and this works well with the time component because once they are > consumed there is a logical grouping of messages from all partitions based > on the time component (coarsely). > The point where I¹m starting to get into trouble is when the consumers are > all stopped for a while and messages start to accumulate in the partitions > without being consumed (hence the lag increases). Once I resume the > consumers (with 1 thread per each partition) messages start to get > consumed very fast, but because some messages take longer to process than > others, over time the lags start to get very uneven between partitions and > this starts to interfere with the grouping by the time component. > So the only way I thought I could prevent this from happening was by > throttling the ³fast² consumers, which have a smaller lag. This will only > happen rarely, so I thought I could live with the approach. To do that, I > obviously need to get the lag information periodically. I do that by a > mechanism similar to what ConsumerOffsetChecker does. But now I need to > know from which thread to call sleep() and there seems to be no decent way > to find that out. > Any better ideas would be highly appreciated. > > Thanks, > Bogdan > > On 08/06/2014 06:25, "Jun Rao" <jun...@gmail.com> wrote: > > >Could you elaborate on the use case of the stream ID? > > > >Thanks, > > > >Jun > > > > > >On Fri, Jun 6, 2014 at 2:13 AM, Bogdan Dimitriu (bdimitri) < > >bdimi...@cisco.com> wrote: > > > >> Hello folks, > >> > >> I¹m using Kafka 0.8.0 with the high level consumer and I have a > >>situation > >> where I need to obtain the ID for each of the KafkaStreams that I > >>create. > >> The KafkaStream class has a method called ³clientId()² that I expected > >> would give me just that, but unfortunately it returns the name of the > >> consumer group. > >> So to make it clear, what I want to obtain is the string that looks like > >> this: myconsumergroup_myhost-1402045464004-2dc0cbf2-0. > >> Is there any way I could get that value for each of the streams? I¹ve > >> looked around the source code but I can¹t see any way to do this. > >> > >> Many thanks, > >> Bogdan > >> > >> > >
