Re: producer behavior when network is down
In Kafka, we detect failures using ZK. So, if the network connectivity btw the producer and the broker is down, but the one btw the broker and ZK is up, we assume the broker is still alive and will continue to send the data to it. Within the the same data center, we assume this is extremely rare. If the network connectivity btw the broker and ZK is also down, the producer will be able to automatically detect the failure and send data to other brokers. In 0.7, there is no way to check whether a message is really sent or not, since the producer requests don't receive any acknowledgement. This is changed in 0.8, where a producer can choose when to receive an ack (see request.required.acks in http://kafka.apache.org/documentation.html#producerconfigs). Thanks, Jun On Fri, Jul 26, 2013 at 9:27 AM, Viktor Kolodrevskiy < viktor.kolodrevs...@gmail.com> wrote: > Hey guys, > > We decided to use Kafka in our new project, now I spend some time to > research how Kafka producer behaves while network connectivity > problems. > > I had 3 virtual machines(ubuntu 13.04, running on Virtualbox) in one > network: > > 1. Kafka server(0.7.2) + Zookeper. > 2. Producer app with default settings. > 3. Consumer app. > > Results of the following tests with default sync producer settings: > > 1. Condition: Put network down on machine (1) for 20 mins. > Result: Producer is working for ~16mins. Consumer does not receive > anything. > After ~16mins Producer app fails(with java.io.IOException: Connection > timed out). Consumer app does not fail. > Messages that were generated during 16mins are lost! > > 2. Condition: Put network down on machine (1) for 5 mins and after 5 > mins start network on (1) again. > Result: Producer app is working, no exceptions or notification that > network was down. > Consumer does not receive messages for 5 mins. But when network on (1) > is up it receives all messages. > There are no messages lost. > > 3. Condition: put network down on machine (2) for 20 mins. > Result: Producer is working for ~16mins. Consumer does not receive > anything. > After ~16mins Producer app fails(with java.io.IOException: Connection > timed out). Consumer app does not fail. > Messages that were generated during 16mins are lost! (Same result as in > test#1) > Kafka and Zookeeper logs that producer is disconnected. > > 4. Condition: Put network down on machine (2) for 5 mins and after 5 > mins start network on (2) again. > Result: Producer app is working, no exceptions or notification that > network was down. > Consumer does not receive messages for 5 mins. But when network on (2) > is up it receives all messages.(Same result as in test#2) > Kafka and Zookeeper logs that producer is disconnected. > > 5. Condition: Kill Kafka server(0.7.2) + Zookeper(kill application, do > not shutdown network). > Result: Producer fails in a few seconds with > "kafka.common.NoBrokersForPartitionException: Partition = null" > Consumer is still working even after 25 minutes. > > One more interesting thing. Changing connect.timeout.ms parameter > value for producer > did not change 16 mins that I have. > > Played with settings and find out the only way to reduce time for > producer to find out that network is down is to change one of two > parameters: reconnect.interval, reconnect.time.interval.ms > > So lets say we change reconnect.time.interval.ms=1000. > This means that producer will do reconnect to kafka every 1 second. > In this case producer find out that network is down in 1 second. > Producer stops sending messages and throw "java.net.ConnectException: > Connection timed out". This is the only way that I found out so far. > In this case we do not loose too much messages but performance may suffer. > Or we can set reconnect.interval=1 so reconnect will happen after each > message sent > and do not loose messages at all. > > Then I did testing for Async producer(producer.type=async) > The results are dramatic for me, coz producer does not throw any exception. > It sends messages and does not fall. > I left it running for night and it did not fall though network between > kafka server and producer app was down. > Playing with async producer config parameters did not help also. > > My questions are: > > 1. Where may these 16 mins come from? > 2. Are there any best practices to handle network down issues? > 3. Why async producer never throws exceptions when network is down? > 4. What is the way to check from sync/async producer that messages > were really sent? >
kafka site facelift
Hey guys, Sriram and I took another pass at the site and documentation: http://kafka.apache.org This includes the following changes: 1. New logo! 2. Get rid of blue header 3. New font (Google webfont, Source Sans Pro) 4. Update the information on the design section to cover 0.8. 5. Move to a single giant page for documentation. The idea here is to gather together all the information into a kind of pseudo manual. This helps trim down the side nav and is hopefully better. 6. Primary documentation now links to 0.8 with a link back to the older docs for those who want it. 7. intro and use case link now link into the uber-documentation page. This is a little weird, not sure what to do about it. There is is still a fair amount left to do: 1. Implementation section needs to be updated for 0.8 2. Tools section should ideally cover the full set of tools Let me know if you see anything broken. Sriram I took another editing pass after your changes. Neha and Jun it would be great if you could review at least the replication section Sriram and I added to see if there is anything you would change. -Jay
Re: Kafka/Hadoop consumers and producers
Kam, I am perfectly fine if you pick this up. After thinking about it for a while, we are going to upgrade to Kafka 0.8.0 and also use Camus as it more closely matches our use case, with the caveat of we do not use Avro. With that said, I will try and work on the back-port of custom data writer patch[1], however, I am not sure how quickly I will get this done as we are going to work towards upgrading our Kafka cluster. Thanks, Andrew [1] https://github.com/linkedin/camus/commit/87917a2aea46da9d21c8f67129f6463af5 2f7aa8 On 8/12/13 6:16 PM, "Kam Kasravi" wrote: >I would like to do this refactoring since I did a high level consumer a >while ago. >A few weeks ago I had opened KAFKA-949 Kafka on Yarn which I was also >hoping to add to contribute. >It's almost done. KAFKA-949 is paired with BIGTOP-989 which adds kafka >0.8 to the bigtop distribution. >KAFKA-949 basically allows kafka brokers to be started up using sysvinit >services and would ease some of the >startup/configuration issues that newbies have when getting started with >kafka. Ideally I would like to >fold a number of kafka/bin/* commands into the kafka service. Andrew >please let me know if would like to >pick this up instead. Thanks! > >Kam > > > > From: Jay Kreps >To: Ken Goodhope >Cc: Andrew Psaltis ; >dibyendu.bhattacha...@pearson.com; "camus_...@googlegroups.com" >; "ao...@wikimedia.org" >; Felix GV ; Cosmin Lehene >; "d...@kafka.apache.org" ; >"users@kafka.apache.org" >Sent: Saturday, August 10, 2013 3:30 PM >Subject: Re: Kafka/Hadoop consumers and producers > > >So guys, just to throw my 2 cents in: > >1. We aren't deprecating anything. I just noticed that the Hadoop contrib >package wasn't getting as much attention as it should. > >2. Andrew or anyone--if there is anyone using the contrib package who >would >be willing to volunteer to kind of adopt it that would be great. I am >happy >to help in whatever way I can. The practical issue is that most of the >committers are either using Camus or not using Hadoop at all so we just >haven't been doing a good job of documenting, bug fixing, and supporting >the contrib packages. > >3. Ken, if you could document how to use Camus that would likely make it a >lot more useful to people. I think most people would want a full-fledged >ETL solution and would likely prefer Camus, but very few people are using >Avro. > >-Jay > > >On Fri, Aug 9, 2013 at 12:27 PM, Ken Goodhope >wrote: > >> I just checked and that patch is in .8 branch. Thanks for working on >> back porting it Andrew. We'd be happy to commit that work to master. >> >> As for the kafka contrib project vs Camus, they are similar but not >>quite >> identical. Camus is intended to be a high throughput ETL for bulk >> ingestion of Kafka data into HDFS. Where as what we have in contrib is >> more of a simple KafkaInputFormat. Neither can really replace the >>other. >> If you had a complex hadoop workflow and wanted to introduce some Kafka >> data into that workflow, using Camus would be a gigantic overkill and a >> pain to setup. On the flipside, if what you want is frequent reliable >> ingest of Kafka data into HDFS, a simple InputFormat doesn't provide you >> with that. >> >> I think it would be preferable to simplify the existing contrib >> Input/OutputFormats by refactoring them to use the more stable higher >>level >> Kafka APIs. Currently they use the lower level APIs. This should make >> them easier to maintain, and user friendly enough to avoid the need for >> extensive documentation. >> >> Ken >> >> >> On Fri, Aug 9, 2013 at 8:52 AM, Andrew Psaltis >>wrote: >> >>> Dibyendu, >>> According to the pull request: >>>https://github.com/linkedin/camus/pull/15it was merged into the >>>camus-kafka-0.8 >>> branch. I have not checked if the code was subsequently removed, >>>however, >>> two at least one the important files from this patch >>>(camus-api/src/main/java/com/linkedin/camus/etl/RecordWriterProvider.jav >>>a) >>> is still present. >>> >>> Thanks, >>> Andrew >>> >>> >>> On Fri, Aug 9, 2013 at 9:39 AM, >>>wrote: >>> Hi Ken, I am also working on making the Camus fit for Non Avro message for our requirement. I see you mentioned about this patch ( https://github.com/linkedin/camus/commit/87917a2aea46da9d21c8f67129f646 3af52f7aa8) which supports custom data writer for Camus. But this patch is not pulled into camus-kafka-0.8 branch. Is there any plan for doing the same ? Regards, Dibyendu -- You received this message because you are subscribed to a topic in the Google Groups "Camus - Kafka ETL for Hadoop" group. To unsubscribe from this topic, visit https://groups.google.com/d/topic/camus_etl/KKS6t5-O-Ng/unsubscribe. To unsubscribe from this group and all its topics, send an email to camus_etl+unsubscr...@googlegroups.com. For more options, visit https://groups.google.com/g
Re: Kafka/Hadoop consumers and producers
We've done a bit of work over at Wikimedia to debianize Kafka and make it behave like a regular service. https://github.com/wikimedia/operations-debs-kafka/blob/debian/debian Most relevant, Ken, is an init script for Kafka: https://github.com/wikimedia/operations-debs-kafka/blob/debian/debian/kafka.init And a bin/kafka shell wrapper for the kafka/bin/*.sh scripts: https://github.com/wikimedia/operations-debs-kafka/blob/debian/debian/bin/kafka I'm about to add an init script for MirrorMaker as well, so mirroring can be demonized and run as a service. On Aug 12, 2013, at 8:16 PM, Kam Kasravi wrote: > I would like to do this refactoring since I did a high level consumer a while > ago. > A few weeks ago I had opened KAFKA-949 Kafka on Yarn which I was also hoping > to add to contribute. > It's almost done. KAFKA-949 is paired with BIGTOP-989 which adds kafka 0.8 to > the bigtop distribution. > KAFKA-949 basically allows kafka brokers to be started up using sysvinit > services and would ease some of the > startup/configuration issues that newbies have when getting started with > kafka. Ideally I would like to > fold a number of kafka/bin/* commands into the kafka service. Andrew please > let me know if would like to > pick this up instead. Thanks! > > Kam > > From: Jay Kreps > To: Ken Goodhope > Cc: Andrew Psaltis ; > dibyendu.bhattacha...@pearson.com; "camus_...@googlegroups.com" > ; "ao...@wikimedia.org" ; > Felix GV ; Cosmin Lehene ; > "d...@kafka.apache.org" ; "users@kafka.apache.org" > > Sent: Saturday, August 10, 2013 3:30 PM > Subject: Re: Kafka/Hadoop consumers and producers > > So guys, just to throw my 2 cents in: > > 1. We aren't deprecating anything. I just noticed that the Hadoop contrib > package wasn't getting as much attention as it should. > > 2. Andrew or anyone--if there is anyone using the contrib package who would > be willing to volunteer to kind of adopt it that would be great. I am happy > to help in whatever way I can. The practical issue is that most of the > committers are either using Camus or not using Hadoop at all so we just > haven't been doing a good job of documenting, bug fixing, and supporting > the contrib packages. > > 3. Ken, if you could document how to use Camus that would likely make it a > lot more useful to people. I think most people would want a full-fledged > ETL solution and would likely prefer Camus, but very few people are using > Avro. > > -Jay > > > On Fri, Aug 9, 2013 at 12:27 PM, Ken Goodhope wrote: > > > I just checked and that patch is in .8 branch. Thanks for working on > > back porting it Andrew. We'd be happy to commit that work to master. > > > > As for the kafka contrib project vs Camus, they are similar but not quite > > identical. Camus is intended to be a high throughput ETL for bulk > > ingestion of Kafka data into HDFS. Where as what we have in contrib is > > more of a simple KafkaInputFormat. Neither can really replace the other. > > If you had a complex hadoop workflow and wanted to introduce some Kafka > > data into that workflow, using Camus would be a gigantic overkill and a > > pain to setup. On the flipside, if what you want is frequent reliable > > ingest of Kafka data into HDFS, a simple InputFormat doesn't provide you > > with that. > > > > I think it would be preferable to simplify the existing contrib > > Input/OutputFormats by refactoring them to use the more stable higher level > > Kafka APIs. Currently they use the lower level APIs. This should make > > them easier to maintain, and user friendly enough to avoid the need for > > extensive documentation. > > > > Ken > > > > > > On Fri, Aug 9, 2013 at 8:52 AM, Andrew Psaltis > > wrote: > > > >> Dibyendu, > >> According to the pull request: https://github.com/linkedin/camus/pull/15it > >> was merged into the camus-kafka-0.8 > >> branch. I have not checked if the code was subsequently removed, however, > >> two at least one the important files from this patch > >> (camus-api/src/main/java/com/linkedin/camus/etl/RecordWriterProvider.java) > >> is still present. > >> > >> Thanks, > >> Andrew > >> > >> > >> On Fri, Aug 9, 2013 at 9:39 AM, wrote: > >> > >>> Hi Ken, > >>> > >>> I am also working on making the Camus fit for Non Avro message for our > >>> requirement. > >>> > >>> I see you mentioned about this patch ( > >>> https://github.com/linkedin/camus/commit/87917a2aea46da9d21c8f67129f6463af52f7aa8) > >>> which supports custom data writer for Camus. But this patch is not pulled > >>> into camus-kafka-0.8 branch. Is there any plan for doing the same ? > >>> > >>> Regards, > >>> Dibyendu > >>> > >>> -- > >>> You received this message because you are subscribed to a topic in the > >>> Google Groups "Camus - Kafka ETL for Hadoop" group. > >>> To unsubscribe from this topic, visit > >>> https://groups.google.com/d/topic/camus_etl/KKS6t5-O-Ng/unsubscribe. > >>> To unsubscribe from this group and all its topics, send an email to > >>>
Re: Kafka/Hadoop consumers and producers
I would like to do this refactoring since I did a high level consumer a while ago. A few weeks ago I had opened KAFKA-949 Kafka on Yarn which I was also hoping to add to contribute. It's almost done. KAFKA-949 is paired with BIGTOP-989 which adds kafka 0.8 to the bigtop distribution. KAFKA-949 basically allows kafka brokers to be started up using sysvinit services and would ease some of the startup/configuration issues that newbies have when getting started with kafka. Ideally I would like to fold a number of kafka/bin/* commands into the kafka service. Andrew please let me know if would like to pick this up instead. Thanks! Kam From: Jay Kreps To: Ken Goodhope Cc: Andrew Psaltis ; dibyendu.bhattacha...@pearson.com; "camus_...@googlegroups.com" ; "ao...@wikimedia.org" ; Felix GV ; Cosmin Lehene ; "d...@kafka.apache.org" ; "users@kafka.apache.org" Sent: Saturday, August 10, 2013 3:30 PM Subject: Re: Kafka/Hadoop consumers and producers So guys, just to throw my 2 cents in: 1. We aren't deprecating anything. I just noticed that the Hadoop contrib package wasn't getting as much attention as it should. 2. Andrew or anyone--if there is anyone using the contrib package who would be willing to volunteer to kind of adopt it that would be great. I am happy to help in whatever way I can. The practical issue is that most of the committers are either using Camus or not using Hadoop at all so we just haven't been doing a good job of documenting, bug fixing, and supporting the contrib packages. 3. Ken, if you could document how to use Camus that would likely make it a lot more useful to people. I think most people would want a full-fledged ETL solution and would likely prefer Camus, but very few people are using Avro. -Jay On Fri, Aug 9, 2013 at 12:27 PM, Ken Goodhope wrote: > I just checked and that patch is in .8 branch. Thanks for working on > back porting it Andrew. We'd be happy to commit that work to master. > > As for the kafka contrib project vs Camus, they are similar but not quite > identical. Camus is intended to be a high throughput ETL for bulk > ingestion of Kafka data into HDFS. Where as what we have in contrib is > more of a simple KafkaInputFormat. Neither can really replace the other. > If you had a complex hadoop workflow and wanted to introduce some Kafka > data into that workflow, using Camus would be a gigantic overkill and a > pain to setup. On the flipside, if what you want is frequent reliable > ingest of Kafka data into HDFS, a simple InputFormat doesn't provide you > with that. > > I think it would be preferable to simplify the existing contrib > Input/OutputFormats by refactoring them to use the more stable higher level > Kafka APIs. Currently they use the lower level APIs. This should make > them easier to maintain, and user friendly enough to avoid the need for > extensive documentation. > > Ken > > > On Fri, Aug 9, 2013 at 8:52 AM, Andrew Psaltis > wrote: > >> Dibyendu, >> According to the pull request: https://github.com/linkedin/camus/pull/15it >> was merged into the camus-kafka-0.8 >> branch. I have not checked if the code was subsequently removed, however, >> two at least one the important files from this patch >> (camus-api/src/main/java/com/linkedin/camus/etl/RecordWriterProvider.java) >> is still present. >> >> Thanks, >> Andrew >> >> >> On Fri, Aug 9, 2013 at 9:39 AM, wrote: >> >>> Hi Ken, >>> >>> I am also working on making the Camus fit for Non Avro message for our >>> requirement. >>> >>> I see you mentioned about this patch ( >>> https://github.com/linkedin/camus/commit/87917a2aea46da9d21c8f67129f6463af52f7aa8) >>> which supports custom data writer for Camus. But this patch is not pulled >>> into camus-kafka-0.8 branch. Is there any plan for doing the same ? >>> >>> Regards, >>> Dibyendu >>> >>> -- >>> You received this message because you are subscribed to a topic in the >>> Google Groups "Camus - Kafka ETL for Hadoop" group. >>> To unsubscribe from this topic, visit >>> https://groups.google.com/d/topic/camus_etl/KKS6t5-O-Ng/unsubscribe. >>> To unsubscribe from this group and all its topics, send an email to >>> camus_etl+unsubscr...@googlegroups.com. >>> For more options, visit https://groups.google.com/groups/opt_out. >>> >> >> -- >> You received this message because you are subscribed to the Google Groups >> "Camus - Kafka ETL for Hadoop" group. >> To unsubscribe from this group and stop receiving emails from it, send an >> email to camus_etl+unsubscr...@googlegroups.com. >> For more options, visit https://groups.google.com/groups/opt_out. >> >> >> > > -- > You received this message because you are subscribed to the Google Groups > "Camus - Kafka ETL for Hadoop" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to camus_etl+unsubscr...@googlegroups.com. > For more options, visit https://groups.google.com/groups/opt_out. >
Re: Kafka 0.7.2 on Scala 2.9
You can package the lib to use another Scala version in 0.7.X project/build.properties Change build.scala.versions=2.8.0 And then ./sbt package To get the lib for your Scala version /*** Joe Stein Founder, Principal Consultant Big Data Open Source Security LLC http://www.stealth.ly Twitter: @allthingshadoop / On Aug 12, 2013, at 7:33 PM, Ian Friedman wrote: > Hi guys, > > Is it possible to compile Kafka 0.7.2 for Scala 2.9? Has anyone done this? > Thanks in advance. > > -- > Ian Friedman >
Kafka 0.7.2 on Scala 2.9
Hi guys, Is it possible to compile Kafka 0.7.2 for Scala 2.9? Has anyone done this? Thanks in advance. -- Ian Friedman
Re: producer behavior when network is down
The goal is to use sync producer and find out that network is down as soon as possible. -- Viktor 2013/8/13 Viktor Kolodrevskiy : > Felix, > the thing is that I was using sync producer. > > -- > Viktor > > 2013/8/13 Felix GV : >> Async production is meant to work this way. You have no delivery guarantee >> nor any exception because the producer sends the message independently of >> the code that called the aync production function. >> >> It is meant to be faster than sync production, but it is obviously intended >> for non-critical messages. >> >> -- >> Felix >> >> >> On Fri, Jul 26, 2013 at 12:27 PM, Viktor Kolodrevskiy < >> viktor.kolodrevs...@gmail.com> wrote: >> >>> Hey guys, >>> >>> We decided to use Kafka in our new project, now I spend some time to >>> research how Kafka producer behaves while network connectivity >>> problems. >>> >>> I had 3 virtual machines(ubuntu 13.04, running on Virtualbox) in one >>> network: >>> >>> 1. Kafka server(0.7.2) + Zookeper. >>> 2. Producer app with default settings. >>> 3. Consumer app. >>> >>> Results of the following tests with default sync producer settings: >>> >>> 1. Condition: Put network down on machine (1) for 20 mins. >>> Result: Producer is working for ~16mins. Consumer does not receive >>> anything. >>> After ~16mins Producer app fails(with java.io.IOException: Connection >>> timed out). Consumer app does not fail. >>> Messages that were generated during 16mins are lost! >>> >>> 2. Condition: Put network down on machine (1) for 5 mins and after 5 >>> mins start network on (1) again. >>> Result: Producer app is working, no exceptions or notification that >>> network was down. >>> Consumer does not receive messages for 5 mins. But when network on (1) >>> is up it receives all messages. >>> There are no messages lost. >>> >>> 3. Condition: put network down on machine (2) for 20 mins. >>> Result: Producer is working for ~16mins. Consumer does not receive >>> anything. >>> After ~16mins Producer app fails(with java.io.IOException: Connection >>> timed out). Consumer app does not fail. >>> Messages that were generated during 16mins are lost! (Same result as in >>> test#1) >>> Kafka and Zookeeper logs that producer is disconnected. >>> >>> 4. Condition: Put network down on machine (2) for 5 mins and after 5 >>> mins start network on (2) again. >>> Result: Producer app is working, no exceptions or notification that >>> network was down. >>> Consumer does not receive messages for 5 mins. But when network on (2) >>> is up it receives all messages.(Same result as in test#2) >>> Kafka and Zookeeper logs that producer is disconnected. >>> >>> 5. Condition: Kill Kafka server(0.7.2) + Zookeper(kill application, do >>> not shutdown network). >>> Result: Producer fails in a few seconds with >>> "kafka.common.NoBrokersForPartitionException: Partition = null" >>> Consumer is still working even after 25 minutes. >>> >>> One more interesting thing. Changing connect.timeout.ms parameter >>> value for producer >>> did not change 16 mins that I have. >>> >>> Played with settings and find out the only way to reduce time for >>> producer to find out that network is down is to change one of two >>> parameters: reconnect.interval, reconnect.time.interval.ms >>> >>> So lets say we change reconnect.time.interval.ms=1000. >>> This means that producer will do reconnect to kafka every 1 second. >>> In this case producer find out that network is down in 1 second. >>> Producer stops sending messages and throw "java.net.ConnectException: >>> Connection timed out". This is the only way that I found out so far. >>> In this case we do not loose too much messages but performance may suffer. >>> Or we can set reconnect.interval=1 so reconnect will happen after each >>> message sent >>> and do not loose messages at all. >>> >>> Then I did testing for Async producer(producer.type=async) >>> The results are dramatic for me, coz producer does not throw any exception. >>> It sends messages and does not fall. >>> I left it running for night and it did not fall though network between >>> kafka server and producer app was down. >>> Playing with async producer config parameters did not help also. >>> >>> My questions are: >>> >>> 1. Where may these 16 mins come from? >>> 2. Are there any best practices to handle network down issues? >>> 3. Why async producer never throws exceptions when network is down? >>> 4. What is the way to check from sync/async producer that messages >>> were really sent? >>> > > > > -- > Thanks, > Viktor -- Thanks, Viktor
Re: producer behavior when network is down
Felix, the thing is that I was using sync producer. -- Viktor 2013/8/13 Felix GV : > Async production is meant to work this way. You have no delivery guarantee > nor any exception because the producer sends the message independently of > the code that called the aync production function. > > It is meant to be faster than sync production, but it is obviously intended > for non-critical messages. > > -- > Felix > > > On Fri, Jul 26, 2013 at 12:27 PM, Viktor Kolodrevskiy < > viktor.kolodrevs...@gmail.com> wrote: > >> Hey guys, >> >> We decided to use Kafka in our new project, now I spend some time to >> research how Kafka producer behaves while network connectivity >> problems. >> >> I had 3 virtual machines(ubuntu 13.04, running on Virtualbox) in one >> network: >> >> 1. Kafka server(0.7.2) + Zookeper. >> 2. Producer app with default settings. >> 3. Consumer app. >> >> Results of the following tests with default sync producer settings: >> >> 1. Condition: Put network down on machine (1) for 20 mins. >> Result: Producer is working for ~16mins. Consumer does not receive >> anything. >> After ~16mins Producer app fails(with java.io.IOException: Connection >> timed out). Consumer app does not fail. >> Messages that were generated during 16mins are lost! >> >> 2. Condition: Put network down on machine (1) for 5 mins and after 5 >> mins start network on (1) again. >> Result: Producer app is working, no exceptions or notification that >> network was down. >> Consumer does not receive messages for 5 mins. But when network on (1) >> is up it receives all messages. >> There are no messages lost. >> >> 3. Condition: put network down on machine (2) for 20 mins. >> Result: Producer is working for ~16mins. Consumer does not receive >> anything. >> After ~16mins Producer app fails(with java.io.IOException: Connection >> timed out). Consumer app does not fail. >> Messages that were generated during 16mins are lost! (Same result as in >> test#1) >> Kafka and Zookeeper logs that producer is disconnected. >> >> 4. Condition: Put network down on machine (2) for 5 mins and after 5 >> mins start network on (2) again. >> Result: Producer app is working, no exceptions or notification that >> network was down. >> Consumer does not receive messages for 5 mins. But when network on (2) >> is up it receives all messages.(Same result as in test#2) >> Kafka and Zookeeper logs that producer is disconnected. >> >> 5. Condition: Kill Kafka server(0.7.2) + Zookeper(kill application, do >> not shutdown network). >> Result: Producer fails in a few seconds with >> "kafka.common.NoBrokersForPartitionException: Partition = null" >> Consumer is still working even after 25 minutes. >> >> One more interesting thing. Changing connect.timeout.ms parameter >> value for producer >> did not change 16 mins that I have. >> >> Played with settings and find out the only way to reduce time for >> producer to find out that network is down is to change one of two >> parameters: reconnect.interval, reconnect.time.interval.ms >> >> So lets say we change reconnect.time.interval.ms=1000. >> This means that producer will do reconnect to kafka every 1 second. >> In this case producer find out that network is down in 1 second. >> Producer stops sending messages and throw "java.net.ConnectException: >> Connection timed out". This is the only way that I found out so far. >> In this case we do not loose too much messages but performance may suffer. >> Or we can set reconnect.interval=1 so reconnect will happen after each >> message sent >> and do not loose messages at all. >> >> Then I did testing for Async producer(producer.type=async) >> The results are dramatic for me, coz producer does not throw any exception. >> It sends messages and does not fall. >> I left it running for night and it did not fall though network between >> kafka server and producer app was down. >> Playing with async producer config parameters did not help also. >> >> My questions are: >> >> 1. Where may these 16 mins come from? >> 2. Are there any best practices to handle network down issues? >> 3. Why async producer never throws exceptions when network is down? >> 4. What is the way to check from sync/async producer that messages >> were really sent? >> -- Thanks, Viktor
Re: producer behavior when network is down
Async production is meant to work this way. You have no delivery guarantee nor any exception because the producer sends the message independently of the code that called the aync production function. It is meant to be faster than sync production, but it is obviously intended for non-critical messages. -- Felix On Fri, Jul 26, 2013 at 12:27 PM, Viktor Kolodrevskiy < viktor.kolodrevs...@gmail.com> wrote: > Hey guys, > > We decided to use Kafka in our new project, now I spend some time to > research how Kafka producer behaves while network connectivity > problems. > > I had 3 virtual machines(ubuntu 13.04, running on Virtualbox) in one > network: > > 1. Kafka server(0.7.2) + Zookeper. > 2. Producer app with default settings. > 3. Consumer app. > > Results of the following tests with default sync producer settings: > > 1. Condition: Put network down on machine (1) for 20 mins. > Result: Producer is working for ~16mins. Consumer does not receive > anything. > After ~16mins Producer app fails(with java.io.IOException: Connection > timed out). Consumer app does not fail. > Messages that were generated during 16mins are lost! > > 2. Condition: Put network down on machine (1) for 5 mins and after 5 > mins start network on (1) again. > Result: Producer app is working, no exceptions or notification that > network was down. > Consumer does not receive messages for 5 mins. But when network on (1) > is up it receives all messages. > There are no messages lost. > > 3. Condition: put network down on machine (2) for 20 mins. > Result: Producer is working for ~16mins. Consumer does not receive > anything. > After ~16mins Producer app fails(with java.io.IOException: Connection > timed out). Consumer app does not fail. > Messages that were generated during 16mins are lost! (Same result as in > test#1) > Kafka and Zookeeper logs that producer is disconnected. > > 4. Condition: Put network down on machine (2) for 5 mins and after 5 > mins start network on (2) again. > Result: Producer app is working, no exceptions or notification that > network was down. > Consumer does not receive messages for 5 mins. But when network on (2) > is up it receives all messages.(Same result as in test#2) > Kafka and Zookeeper logs that producer is disconnected. > > 5. Condition: Kill Kafka server(0.7.2) + Zookeper(kill application, do > not shutdown network). > Result: Producer fails in a few seconds with > "kafka.common.NoBrokersForPartitionException: Partition = null" > Consumer is still working even after 25 minutes. > > One more interesting thing. Changing connect.timeout.ms parameter > value for producer > did not change 16 mins that I have. > > Played with settings and find out the only way to reduce time for > producer to find out that network is down is to change one of two > parameters: reconnect.interval, reconnect.time.interval.ms > > So lets say we change reconnect.time.interval.ms=1000. > This means that producer will do reconnect to kafka every 1 second. > In this case producer find out that network is down in 1 second. > Producer stops sending messages and throw "java.net.ConnectException: > Connection timed out". This is the only way that I found out so far. > In this case we do not loose too much messages but performance may suffer. > Or we can set reconnect.interval=1 so reconnect will happen after each > message sent > and do not loose messages at all. > > Then I did testing for Async producer(producer.type=async) > The results are dramatic for me, coz producer does not throw any exception. > It sends messages and does not fall. > I left it running for night and it did not fall though network between > kafka server and producer app was down. > Playing with async producer config parameters did not help also. > > My questions are: > > 1. Where may these 16 mins come from? > 2. Are there any best practices to handle network down issues? > 3. Why async producer never throws exceptions when network is down? > 4. What is the way to check from sync/async producer that messages > were really sent? >
Re: trouble building 0.8
That's a classic: just delete your ~/.ivy2 as well and be prepared to
download the internet again :(
Erik.
Op 06-08-13 05:36, Rob Withers schreef:
I deleted all the sbt project and target stuff, for both ~/.sbt and
kafka/project/build. I had previously had php stuff in my global sbt stuff.
This resolved this issue, but now I am having another…
I get the following:
[warn] Multiple resolvers having different access mechanism configured with
same name 'sbt-plugin-releases'. To avoid conflict, Remove duplicate project
resolvers (`resolvers`) or rename publishing resolver (`publishTo`).
I also get this failure to resolve dependency:
[info] Resolving org.scalatest#scalatest_2.10;1.8 ...
[warn] module not found: org.scalatest#scalatest_2.10;1.8
[warn] local: tried
[warn]
/Users/reefedjib/.ivy2/local/org.scalatest/scalatest_2.10/1.8/ivys/ivy.xml
[warn] SonaType ScalaTest repo: tried
[warn]
https://oss.sonatype.org/content/groups/public/org/scalatest/org/scalatest/scalatest_2.10/1.8/scalatest_2.10-1.8.pom
[warn] public: tried
[warn]
http://repo1.maven.org/maven2/org/scalatest/scalatest_2.10/1.8/scalatest_2.10-1.8.pom
I looked at:
http://repo1.maven.org/maven2/org/scalatest/scalatest_2.10
and there is no 1.8, but a 1.9…so, the issue is in core/build.sbt, where scalatest should
be "1.9.1". I change it and it works. This was changed today, so perhaps they
lost 1.8?
thanks,
rob
On Aug 5, 2013, at 8:19 PM, Rob Withers wrote:
Well, I changed something, as it was working yesterday. Here's my attempt at
updating…
Robs-MacBook-Pro:kafka reefedjib$ sbt "++2.10.2 update"
[info] Loading global plugins from /Users/reefedjib/.sbt/plugins
[info] Loading project definition from
/Users/reefedjib/Desktop/rob/comp/workspace-frameworks/kafka/project
[warn] Multiple resolvers having different access mechanism configured with
same name 'sbt-plugin-releases'. To avoid conflict, Remove duplicate project
resolvers (`resolvers`) or rename publishing resolver (`publishTo`).
[error] AttributeKey ID collisions detected for: 'pgp-signer'
(sbt.Task[com.jsuereth.pgp.sbtplugin.PgpSigner],
sbt.Task[com.typesafe.sbt.pgp.PgpSigner]), 'pgp-verifier'
(sbt.Task[com.jsuereth.pgp.sbtplugin.PgpVerifier],
sbt.Task[com.typesafe.sbt.pgp.PgpVerifier]), 'check-pgp-signatures'
(sbt.Task[com.typesafe.sbt.pgp.SignatureCheckReport],
sbt.Task[com.jsuereth.pgp.sbtplugin.SignatureCheckReport]), 'signatures-module'
(sbt.Task[com.typesafe.sbt.pgp.GetSignaturesModule],
sbt.Task[com.jsuereth.pgp.sbtplugin.GetSignaturesModule])
[error] Use 'last' for the full log.
So here's my /Users/reefedjib/.sbt/plugins:
resolvers += Classpaths.typesafeResolver
addSbtPlugin("com.typesafe.sbteclipse" % "sbteclipse-plugin" % "2.2.0")
and I just clone 0.8, no changes.
Is there anywhere else I need to look for sbt plugin configs, outside akka?
thanks,
rob
--
Erik van Oosten
http://www.day-to-day-stuff.blogspot.com/
Re: Error messages in logs
The closed by interrupt exception is because the underlying "leader-finder-thread" has been interrupted - this is done for e.g., on a rebalance. Do you see any rebalancing messages in your consumer logs? How many topics are you consuming? Are the topics present on the broker and is there a leader for each of them (i.e., without any leader movement)? Can you try running a console consumer to see if you see the same behavior? Thanks, Joel On Mon, Aug 12, 2013 at 10:27 AM, Eric Sites wrote: > Very clean logs. > > Just deletes and rolls messages: > > [2013-08-12 16:49:15,371] INFO [Kafka Log on Broker 4], Rolling incoming-6 > due to full data log (kafka.log.Log) > [2013-08-12 16:49:15,429] INFO [Kafka Log on Broker 4], Rolling log > 'incoming-6' to /mnt/data/kafka/incoming-6/00434346.log and > /mnt/data/kafka/incoming-6/00434346.index (kafka.log.Log) > > [2013-08-12 17:18:59,635] INFO [Kafka Log on Broker 4], Deleting log > segment 63488 from incoming-1 (kafka.log.Log) > [2013-08-12 17:18:59,636] INFO Deleting index > /mnt/data/kafka/incoming-1/00063488.index > (kafka.log.OffsetIndex) > > [2013-08-12 17:19:59,635] INFO [Kafka Log on Broker 4], Deleting log > segment 11276 from incoming-4 (kafka.log.Log) > [2013-08-12 17:19:59,635] INFO Deleting index > /mnt/data/kafka/incoming-4/00011276.index > (kafka.log.OffsetIndex) > [2013-08-12 17:19:59,636] INFO [Kafka Log on Broker 4], Deleting log > segment 63924 from incoming-6 (kafka.log.Log) > [2013-08-12 17:19:59,637] INFO Deleting index > /mnt/data/kafka/incoming-6/00063924.index > (kafka.log.OffsetIndex) > > [2013-08-12 17:19:59,635] INFO [Kafka Log on Broker 4], Deleting log > segment 11276 from incoming-4 (kafka.log.Log) > [2013-08-12 17:19:59,635] INFO Deleting index > /mnt/data/kafka/incoming-4/00011276.index > (kafka.log.OffsetIndex) > [2013-08-12 17:19:59,636] INFO [Kafka Log on Broker 4], Deleting log > segment 63924 from incoming-6 (kafka.log.Log) > [2013-08-12 17:19:59,637] INFO Deleting index > /mnt/data/kafka/incoming-6/00063924.index > (kafka.log.OffsetIndex) > > [2013-08-12 17:20:59,635] INFO [Kafka Log on Broker 4], Deleting log > segment 64723 from incoming-0 (kafka.log.Log) > [2013-08-12 17:20:59,636] INFO Deleting index > /mnt/data/kafka/incoming-0/00064723.index > (kafka.log.OffsetIndex) > > [2013-08-12 17:20:59,635] INFO [Kafka Log on Broker 4], Deleting log > segment 64723 from incoming-0 (kafka.log.Log) > [2013-08-12 17:20:59,636] INFO Deleting index > /mnt/data/kafka/incoming-0/00064723.index > (kafka.log.OffsetIndex) > > [2013-08-12 17:23:59,636] INFO [Kafka Log on Broker 4], Deleting log > segment 65478 from incoming-1 (kafka.log.Log) > [2013-08-12 17:23:59,637] INFO Deleting index > /mnt/data/kafka/incoming-1/00065478.index > (kafka.log.OffsetIndex) > > > Cheers, > Eric Sites > > On 8/12/13 1:20 PM, "Joel Koshy" wrote: > >>The consumer uses a sync.producer to look up topic metadata for topics >>that it consumes. For some reason, it is unable to establish a >>connection to the brokers. Are the brokers all up with clean logs? >> >>On Mon, Aug 12, 2013 at 8:39 AM, Eric Sites >>wrote: >>> I am getting some new messages in my 0.8 Java producer app. >>> >>> I am running 6 Kafka brokers. >>> >>> The app producing these logs messages is not doing anything, i.e. Not >>>producing messages. >>> >>> Anywhere form every 5 minutes to 10 seconds sometimes. >>> >>> [2013-08-12 14:45:20,714] ERROR Producer connection to kafka002:9092 >>>unsuccessful (kafka.producer.SyncProducer) >>> java.nio.channels.ClosedByInterruptException >>> at >>>java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterrupti >>>bleChannel.java:184) >>> at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) >>> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) >>> at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) >>> at >>>kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) >>> at >>>kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProdu >>>cer.scala:68) >>> at kafka.producer.SyncProducer.send(SyncProducer.scala:112) >>> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) >>> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) >>> at >>>kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerF >>>etcherManager.scala:64) >>> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) >>> [2013-08-12 14:45:20,714] ERROR Producer connection to kafka006:9092 >>>unsuccessful (kafka.producer.SyncProducer) >>> java.nio.channels.ClosedByInterruptException >>> at >>>java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterrupti >>>bleChannel.java:184) >>> at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) >>> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) >>> at kafka.pr
Re: Error messages in logs
Very clean logs. Just deletes and rolls messages: [2013-08-12 16:49:15,371] INFO [Kafka Log on Broker 4], Rolling incoming-6 due to full data log (kafka.log.Log) [2013-08-12 16:49:15,429] INFO [Kafka Log on Broker 4], Rolling log 'incoming-6' to /mnt/data/kafka/incoming-6/00434346.log and /mnt/data/kafka/incoming-6/00434346.index (kafka.log.Log) [2013-08-12 17:18:59,635] INFO [Kafka Log on Broker 4], Deleting log segment 63488 from incoming-1 (kafka.log.Log) [2013-08-12 17:18:59,636] INFO Deleting index /mnt/data/kafka/incoming-1/00063488.index (kafka.log.OffsetIndex) [2013-08-12 17:19:59,635] INFO [Kafka Log on Broker 4], Deleting log segment 11276 from incoming-4 (kafka.log.Log) [2013-08-12 17:19:59,635] INFO Deleting index /mnt/data/kafka/incoming-4/00011276.index (kafka.log.OffsetIndex) [2013-08-12 17:19:59,636] INFO [Kafka Log on Broker 4], Deleting log segment 63924 from incoming-6 (kafka.log.Log) [2013-08-12 17:19:59,637] INFO Deleting index /mnt/data/kafka/incoming-6/00063924.index (kafka.log.OffsetIndex) [2013-08-12 17:19:59,635] INFO [Kafka Log on Broker 4], Deleting log segment 11276 from incoming-4 (kafka.log.Log) [2013-08-12 17:19:59,635] INFO Deleting index /mnt/data/kafka/incoming-4/00011276.index (kafka.log.OffsetIndex) [2013-08-12 17:19:59,636] INFO [Kafka Log on Broker 4], Deleting log segment 63924 from incoming-6 (kafka.log.Log) [2013-08-12 17:19:59,637] INFO Deleting index /mnt/data/kafka/incoming-6/00063924.index (kafka.log.OffsetIndex) [2013-08-12 17:20:59,635] INFO [Kafka Log on Broker 4], Deleting log segment 64723 from incoming-0 (kafka.log.Log) [2013-08-12 17:20:59,636] INFO Deleting index /mnt/data/kafka/incoming-0/00064723.index (kafka.log.OffsetIndex) [2013-08-12 17:20:59,635] INFO [Kafka Log on Broker 4], Deleting log segment 64723 from incoming-0 (kafka.log.Log) [2013-08-12 17:20:59,636] INFO Deleting index /mnt/data/kafka/incoming-0/00064723.index (kafka.log.OffsetIndex) [2013-08-12 17:23:59,636] INFO [Kafka Log on Broker 4], Deleting log segment 65478 from incoming-1 (kafka.log.Log) [2013-08-12 17:23:59,637] INFO Deleting index /mnt/data/kafka/incoming-1/00065478.index (kafka.log.OffsetIndex) Cheers, Eric Sites On 8/12/13 1:20 PM, "Joel Koshy" wrote: >The consumer uses a sync.producer to look up topic metadata for topics >that it consumes. For some reason, it is unable to establish a >connection to the brokers. Are the brokers all up with clean logs? > >On Mon, Aug 12, 2013 at 8:39 AM, Eric Sites >wrote: >> I am getting some new messages in my 0.8 Java producer app. >> >> I am running 6 Kafka brokers. >> >> The app producing these logs messages is not doing anything, i.e. Not >>producing messages. >> >> Anywhere form every 5 minutes to 10 seconds sometimes. >> >> [2013-08-12 14:45:20,714] ERROR Producer connection to kafka002:9092 >>unsuccessful (kafka.producer.SyncProducer) >> java.nio.channels.ClosedByInterruptException >> at >>java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterrupti >>bleChannel.java:184) >> at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) >> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) >> at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) >> at >>kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) >> at >>kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProdu >>cer.scala:68) >> at kafka.producer.SyncProducer.send(SyncProducer.scala:112) >> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) >> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) >> at >>kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerF >>etcherManager.scala:64) >> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) >> [2013-08-12 14:45:20,714] ERROR Producer connection to kafka006:9092 >>unsuccessful (kafka.producer.SyncProducer) >> java.nio.channels.ClosedByInterruptException >> at >>java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterrupti >>bleChannel.java:184) >> at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) >> at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) >> at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) >> at >>kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) >> at >>kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProdu >>cer.scala:68) >> at kafka.producer.SyncProducer.send(SyncProducer.scala:112) >> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) >> at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) >> at >>kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerF >>etcherManager.scala:64) >> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) >> [2013-08-12 14:45:20,715] E
Re: Error messages in logs
The consumer uses a sync.producer to look up topic metadata for topics that it consumes. For some reason, it is unable to establish a connection to the brokers. Are the brokers all up with clean logs? On Mon, Aug 12, 2013 at 8:39 AM, Eric Sites wrote: > I am getting some new messages in my 0.8 Java producer app. > > I am running 6 Kafka brokers. > > The app producing these logs messages is not doing anything, i.e. Not > producing messages. > > Anywhere form every 5 minutes to 10 seconds sometimes. > > [2013-08-12 14:45:20,714] ERROR Producer connection to kafka002:9092 > unsuccessful (kafka.producer.SyncProducer) > java.nio.channels.ClosedByInterruptException > at > java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) > at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) > at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) > at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) > at > kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) > at kafka.producer.SyncProducer.send(SyncProducer.scala:112) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) > at > kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) > at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) > [2013-08-12 14:45:20,714] ERROR Producer connection to kafka006:9092 > unsuccessful (kafka.producer.SyncProducer) > java.nio.channels.ClosedByInterruptException > at > java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) > at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) > at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) > at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) > at > kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) > at kafka.producer.SyncProducer.send(SyncProducer.scala:112) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) > at > kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) > at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) > [2013-08-12 14:45:20,715] ERROR Producer connection to kafka001:9092 > unsuccessful (kafka.producer.SyncProducer) > java.nio.channels.ClosedByInterruptException > at > java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) > at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) > at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) > at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) > at > kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) > at kafka.producer.SyncProducer.send(SyncProducer.scala:112) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) > at > kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) > at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) > [2013-08-12 14:45:20,715] ERROR Producer connection to kafka004:9092 > unsuccessful (kafka.producer.SyncProducer) > java.nio.channels.ClosedByInterruptException > at > java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) > at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) > at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) > at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) > at > kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) > at kafka.producer.SyncProducer.send(SyncProducer.scala:112) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) > at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) > at > kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) > at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) > [2013-08-12 14:45:20,716] ERROR Producer connection to kafka005:9092 > unsuccessful (kafka.producer.SyncProducer) > java.nio.channels.ClosedByInterruptException > at > java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) > at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) > at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57
Error messages in logs
I am getting some new messages in my 0.8 Java producer app. I am running 6 Kafka brokers. The app producing these logs messages is not doing anything, i.e. Not producing messages. Anywhere form every 5 minutes to 10 seconds sometimes. [2013-08-12 14:45:20,714] ERROR Producer connection to kafka002:9092 unsuccessful (kafka.producer.SyncProducer) java.nio.channels.ClosedByInterruptException at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) at kafka.producer.SyncProducer.send(SyncProducer.scala:112) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) [2013-08-12 14:45:20,714] ERROR Producer connection to kafka006:9092 unsuccessful (kafka.producer.SyncProducer) java.nio.channels.ClosedByInterruptException at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) at kafka.producer.SyncProducer.send(SyncProducer.scala:112) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) [2013-08-12 14:45:20,715] ERROR Producer connection to kafka001:9092 unsuccessful (kafka.producer.SyncProducer) java.nio.channels.ClosedByInterruptException at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) at kafka.producer.SyncProducer.send(SyncProducer.scala:112) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) [2013-08-12 14:45:20,715] ERROR Producer connection to kafka004:9092 unsuccessful (kafka.producer.SyncProducer) java.nio.channels.ClosedByInterruptException at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) at kafka.producer.SyncProducer.send(SyncProducer.scala:112) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:88) at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:64) at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:51) [2013-08-12 14:45:20,716] ERROR Producer connection to kafka005:9092 unsuccessful (kafka.producer.SyncProducer) java.nio.channels.ClosedByInterruptException at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184) at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:543) at kafka.network.BlockingChannel.connect(BlockingChannel.scala:57) at kafka.producer.SyncProducer.connect(SyncProducer.scala:146) at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:161) at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:68) at kafka.producer.SyncProducer.send(SyncProducer.scala:112) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53) at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUti
Re: Kafka 08 clients
This is the Kafka C client for 0.8 we are using at Wikimedia:
https://github.com/edenhill/librdkafka
If you're using Debian/Ubuntu: you use the debian branch here to build a .deb:
https://github.com/paravoid/librdkafka/tree/debian
On Aug 12, 2013, at 12:06 AM, Jun Rao wrote:
> At LinkedIn, we built a C producer client for 0.8, we plan to open source
> it in the next few weeks.
>
> Thanks,
>
> Jun
>
>
> On Sat, Aug 10, 2013 at 6:29 PM, Mark wrote:
>
>> Is there an *official* client out there?
>>
>> On Aug 10, 2013, at 4:10 PM, Scott Clasen wrote:
>>
>>> bpot/poseidon on github is a ruby 0.8 client, works fine for me
>>>
>>> Sent from my iPhone
>>>
>>> On Aug 10, 2013, at 3:08 PM, Timothy Chen wrote:
>>>
That's definitely means it's not up to date to the protocol, I'm tried
>> the
java client and it was working with latest 0.8 api.
Not sure about any other languages.
Tim
On Sat, Aug 10, 2013 at 2:55 PM, Mark
>> wrote:
> Are all Kafka clients working with the latest version of Kafka?
>
> I tried the kafka-rb client and a simple example listed in the README
>> but
> I keep getting a nasty error
> require 'kafka'
> producer = Kafka::Producer.new
> message = Kafka::Message.new("some random message content")
> producer.push(message)
>
> [2013-08-10 14:49:52,166] ERROR Closing socket for /127.0.0.1 because
>> of
> error (kafka.network.Processor)
> java.nio.BufferUnderflowException
> at java.nio.HeapByteBuffer.get(HeapByteBuffer.java:127)
> at java.nio.ByteBuffer.get(ByteBuffer.java:675)
> at kafka.api.ApiUtils$.readShortString(ApiUtils.scala:38)
> at
> kafka.api.ProducerRequest$$anonfun$1.apply(ProducerRequest.scala:40)
> at
> kafka.api.ProducerRequest$$anonfun$1.apply(ProducerRequest.scala:38)
> at
>
>> scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:227)
> at
>
>> scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:227)
> at
> scala.collection.immutable.Range$ByOne$class.foreach(Range.scala:282)
> at
> scala.collection.immutable.Range$$anon$1.foreach(Range.scala:274)
> at
>
>> scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:227)
> at scala.collection.immutable.Range.flatMap(Range.scala:39)
> at kafka.api.ProducerRequest$.readFrom(ProducerRequest.scala:38)
> at kafka.api.RequestKeys$$anonfun$1.apply(RequestKeys.scala:34)
> at kafka.api.RequestKeys$$anonfun$1.apply(RequestKeys.scala:34)
> at
> kafka.network.RequestChannel$Request.(RequestChannel.scala:49)
> at kafka.network.Processor.read(SocketServer.scala:345)
> at kafka.network.Processor.run(SocketServer.scala:245)
> at java.lang.Thread.run(Thread.java:680)
>
>
>>
>>
