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.
The kafka stream does not have an ID itself, the one you mentioned is the ID of the fetcher thread that put data into the stream. Although there is a one-to-one mapping between the fetcher thread and the stream, the Ids of the fetcher cannot be accessed from the kafka stream itself.
Guozhang On Fri, Jun 6, 2014 at 2:13 AM, Bogdan Dimitriu (bdimitri) < [EMAIL PROTECTED]> wrote:
I see. And can I somehow reliably get the ID of the fetcher thread that is providing data for a KafkaStream? Specifically I¹d like to know that ID from the thread where I¹m consuming (where I iterate through the stream).
Thank you, Bogdan
On 06/06/2014 16:16, "Guozhang Wang" <[EMAIL PROTECTED]> 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.
On 08/06/2014 06:25, "Jun Rao" <[EMAIL PROTECTED]> wrote:
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) < [EMAIL PROTECTED]> wrote:
Which JMX MBeans are you referring to, Jun? I couldn¡¯t find anything that gives me the same information as the ConsumerOffsetChecker tool. In any case, my main problem is that I don¡¯t know when I should slow down the iteration because I don¡¯t know which stream the iteration is consuming. I have the global partition lag situation (from the ConsumerOffsetChecker), but I don¡¯t know how to apply it to slow down the iterators because I can¡¯t identify them.
On 09/06/2014 15:28, "Jun Rao" <[EMAIL PROTECTED]> wrote: