Re: Getting the KafkaStream ID
The jmx should be of the form of clientId*-ConsumerLag under kafka.server. Pausing the iteration will indirectly pause the underlying fetcher. Thanks, Jun On Wed, Jun 11, 2014 at 3:09 AM, Bogdan Dimitriu (bdimitri) < bdimi...@cisco.com> 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. > > Thanks, > Bogdan > > On 09/06/2014 15:28, "Jun Rao" wrote: > > >We do have a jmx that reports the lag per partition. You could probably > >get > >the lag that way. Then, you just need to slow down the iteration on the > >fast partition. > > > >Thanks, > > > >Jun > > > > > >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" 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 > >> >> > >> >> > >> > >> > >
Re: Getting the KafkaStream ID
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. Thanks, Bogdan On 09/06/2014 15:28, "Jun Rao" wrote: >We do have a jmx that reports the lag per partition. You could probably >get >the lag that way. Then, you just need to slow down the iteration on the >fast partition. > >Thanks, > >Jun > > >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" 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 >> >> >> >> >> >>
Re: Getting the KafkaStream ID
We do have a jmx that reports the lag per partition. You could probably get the lag that way. Then, you just need to slow down the iteration on the fast partition. Thanks, Jun 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" 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 > >> > >> > >
Re: Getting the KafkaStream ID
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" 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 > >> > >> > >
Re: Getting the KafkaStream ID
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" 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 >> >>
Re: Getting the KafkaStream ID
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 > >
Re: Getting the KafkaStream ID
It is not easy to get the fetcher thread id from kafka stream. We do have the mapping as a topicThreadIdAndQueues, but it is private to the consumer and not exposed to callers. Guozhang On Fri, Jun 6, 2014 at 9:00 AM, Bogdan Dimitriu (bdimitri) < bdimi...@cisco.com> 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" wrote: > > >Bogdan, > > > >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) < > >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 > >> > >> > > > > > >-- > >-- Guozhang > > -- -- Guozhang
Re: Getting the KafkaStream ID
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" wrote: >Bogdan, > >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) < >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 >> >> > > >-- >-- Guozhang
Re: Getting the KafkaStream ID
Bogdan, 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) < 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 > > -- -- Guozhang