Re: Producer fails to send data when it is used in log4j appender.

[Jun Rao]

I think it tries to add the logging in Kafka itself back to the KafkaAppender.
This creates an infinite loop. Maybe you could try setting the log level in
Kafka package to OFF?

Thanks,

Jun


On Mon, Mar 3, 2014 at 6:26 PM, 김동경  wrote:

> I made simple log4j kafka appender.
> I copied most of the code from 0.8.0 Producer example in "
> https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example";
> to code "append" function.
>
> I confirmed producer example code is working with my environment.
> But when I use same logic for log4j appender, it didn`t work.
> It is trying to fetch metadata repeatedly and I am getting infinite
> "Utils$.swallowError" error.
>
> I have no idea on swallowError.
> It looks it failed to fetch metadata from broker, it is trying again and
> again.
> Max retries count is just 3, but I don`t know why it happens.
>
> Are there anything that should be done to produce log data into Kafka via
> log4j Appender?
>
>
> -
> INFO [main] (Logging.scala:67) - Verifying properties
>  INFO [main] (Logging.scala:67) - Property metadata.broker.list is
> overridden to kafka01:9092
>  WARN [main] (Logging.scala:82) - Property zk.connect is not valid
>  INFO [main] (Logging.scala:67) - Property request.required.acks is
> overridden to 1
>  INFO [main] (Logging.scala:67) - Property partitioner.class is overridden
> to com.samsung.rtdp.SimplePartitioner2
>  INFO [main] (Logging.scala:67) - Property serializer.class is overridden
> to kafka.serializer.StringEncoder
>  INFO [main] (HelloWorld.java:14) - Entering application.
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 0 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 1 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 2 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 3 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 4 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 5 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 6 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 7 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 8 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 9 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 10 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 11 for 1 topic(s)
> Set(KafkaAppenderTest)
>  INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
> id:0,host:kafka01,port:9092 with correlation id 12 for 1 topic(s)
> Set(KafkaAppenderTest)
> .
> .
> .
> java.lang.StackOverflowError
> at java.lang.StringCoding.deref(StringCoding.java:64)
> at java.lang.StringCoding.encode(StringCoding.java:275)
> at java.lang.String.getBytes(String.java:954)
> at java.io.UnixFileSystem.getBooleanAttributes0(Native Method)
> at java.io.UnixFileSystem.getBooleanAttributes(UnixFileSystem.java:243)
> at java.io.File.exists(File.java:791)
> at sun.misc.URLClassPath$FileLoader.getResource(URLClassPath.java:1014)
> at sun.misc.URLClassPath.getResource(URLClassPath.java:189)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:209)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:205)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:323)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:268)
> at java.lang.ClassLoader.defineClass1(Native Method)
> at java.lang.ClassLoader.defineClass(ClassLoader.java:643)
> at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
> at java.net.URLClassLoader.defineClass(URLClassLoader.java:277)
> at java.net.URLClassLoader.access$000(URLClassLoader.java:73)
> at 

Re: Consumer group ID for high level consumer

[Binita Bharati]

Hi Neha,


I had to set the "auto.offset.reset" to "smallest" in the particular case
where consumer threads weren't running when the data was produced.

Then on, its working as you said, ie , as long as the group.id is same, the
consumer threads continue to receive new data.

Thanks


On Mon, Mar 3, 2014 at 10:55 PM, Neha Narkhede wrote:

> Question  : How to make the consumer threads receive the older data that
> was produced when the consumer threads aren't running ?
>
> As long as you make sure the group.id doesn't change, the consumer will
> always
> consume data from where it left off during its last run. In other words,
> the consumer
> can consume data that was produced when it wasn't running, if it is always
> started
> using the same group.id
>
> Thanks,
> Neha
>
>
> On Mon, Mar 3, 2014 at 8:02 AM, Binita Bharati  >wrote:
>
> > Hi all,
> >
> > The HighLevelConsumer is working fine. Ref :
> > https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
> .
> >
> > I was expecting the consumer threads to be able to receive older data (
> ie
> > the data put into the topic when the consumer threads weren't running )
> > That was the root of the problem, and once, I started my consumer
> > application first, and then, generated fresh data into the topic, the
> > consumer threads started receiving the new data.
> >
> > Question  : How to make the consumer threads receive the older data that
> > was produced when the consumer threads aren't running ?
> >
> > Thanks
> >
> >
> > On Wed, Feb 26, 2014 at 6:30 PM, Martin Kleppmann
> > wrote:
> >
> > > Hi Binita,
> > >
> > > The consumer group (group.id) is a mechanism for sharing the load of
> > > consuming a high-volume topic between multiple consumers. If you don't
> > set
> > > a group ID, each consumer consumes all the partitions of a topic. If
> you
> > > set several consumers to the same group ID, the partitions of the topic
> > > you're consuming will be shared amongst the consumers in that group, so
> > > that each message is delivered to only one of the consumers in the
> group.
> > >
> > > You can set the group ID to be whatever you like.
> > >
> > > Hope that helps,
> > > Martin
> > >
> > > On 25 Feb 2014, at 06:27, Binita Bharati 
> > wrote:
> > > > Hi all,
> > > >
> > > > I am referring to this e.g:
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
> > > .
> > > >
> > > > What is the consumer group ID being referred here ?
> > > >
> > > > Thanks
> > > > Binita
> > >
> > >
> >
>

Producer fails to send data when it is used in log4j appender.

[김동경]

I made simple log4j kafka appender.
I copied most of the code from 0.8.0 Producer example in "
https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+Producer+Example";
to code "append" function.

I confirmed producer example code is working with my environment.
But when I use same logic for log4j appender, it didn`t work.
It is trying to fetch metadata repeatedly and I am getting infinite
"Utils$.swallowError" error.

I have no idea on swallowError.
It looks it failed to fetch metadata from broker, it is trying again and
again.
Max retries count is just 3, but I don`t know why it happens.

Are there anything that should be done to produce log data into Kafka via
log4j Appender?

-
INFO [main] (Logging.scala:67) - Verifying properties
 INFO [main] (Logging.scala:67) - Property metadata.broker.list is
overridden to kafka01:9092
 WARN [main] (Logging.scala:82) - Property zk.connect is not valid
 INFO [main] (Logging.scala:67) - Property request.required.acks is
overridden to 1
 INFO [main] (Logging.scala:67) - Property partitioner.class is overridden
to com.samsung.rtdp.SimplePartitioner2
 INFO [main] (Logging.scala:67) - Property serializer.class is overridden
to kafka.serializer.StringEncoder
 INFO [main] (HelloWorld.java:14) - Entering application.
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 0 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 1 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 2 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 3 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 4 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 5 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 6 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 7 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 8 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 9 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 10 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 11 for 1 topic(s)
Set(KafkaAppenderTest)
 INFO [main] (HelloWorld.java:14) - Fetching metadata from broker
id:0,host:kafka01,port:9092 with correlation id 12 for 1 topic(s)
Set(KafkaAppenderTest)
.
.
.
java.lang.StackOverflowError
at java.lang.StringCoding.deref(StringCoding.java:64)
at java.lang.StringCoding.encode(StringCoding.java:275)
at java.lang.String.getBytes(String.java:954)
at java.io.UnixFileSystem.getBooleanAttributes0(Native Method)
at java.io.UnixFileSystem.getBooleanAttributes(UnixFileSystem.java:243)
at java.io.File.exists(File.java:791)
at sun.misc.URLClassPath$FileLoader.getResource(URLClassPath.java:1014)
at sun.misc.URLClassPath.getResource(URLClassPath.java:189)
at java.net.URLClassLoader$1.run(URLClassLoader.java:209)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:205)
at java.lang.ClassLoader.loadClass(ClassLoader.java:323)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294)
at java.lang.ClassLoader.loadClass(ClassLoader.java:268)
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(ClassLoader.java:643)
at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
at java.net.URLClassLoader.defineClass(URLClassLoader.java:277)
at java.net.URLClassLoader.access$000(URLClassLoader.java:73)
at java.net.URLClassLoader$1.run(URLClassLoader.java:212)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:205)
at java.lang.ClassLoader.loadClass(ClassLoader.java:323)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294)
at java.lang.ClassLoader.loadClass(ClassLoader.java:268)
at
org.apache.log4j.spi.ThrowableInformation.getThrowableStrRep(

Re: can't connect to kafka from a java client.

[김동경]

Just for your information, I faced same issue.
For me, it was origin from the hostname of Kafka node.

Since the hostname of broker stored in Zookeeper was different with the
hostname what I configured in source,
it couldn`t fetch the meta data of brokers.

If you configure the log level as debug, and look into zookeeper or broker
log, you can easily find the hostname stored in zookeeper.

Thanks
Dongkyoung


2014-02-28 0:41 GMT+09:00 Jun Rao :

> zk.connect is needed in Kafka 0.7. Since you are using Kafka 0.8, you need
> to set metadata.broker.list. ZK is no longer needed in the producer in 0.8.
> You can follow http://kafka.apache.org/documentation.html for 0.8
> documentation.
>
> The latest release in maven is 0.8.0. We publish different kafka jars for
> different versions of scala. That's why you see 2.8.0, 2.10.0, etc. Those
> refer to the scala versions. There is a problem with kafka_2.8.0-0.8.0 in
> maven (we will fix it in the next release). The rest of the scala versions
> are fine.
>
> Let us know if you see any other problems.
>
> Thanks,
>
> Jun
>
>
> On Thu, Feb 27, 2014 at 3:32 AM, rails  wrote:
>
> > Goal: I am trying to send messages to kafka from a java cleint.
> > And it has been a pain..
> > Let me describe in brief.
> >
> > 1. I have installed kafka on a centos VM.
> > 2. I ran the zookeeper that comes with it, the server , the producer and
> > the client with all of the default properties files.
> > I sent and received messages successfully.
> > 3. I have a telnet connection from my computer to the zookeeper (2181)
> port
> > and the kafka server (9092) port at the VM.
> >
> > Now, I want write java code to send messages to the topic.
> > I use the example from the quick start at the site:
> >
> > Properties props = new Properties();
> > props.put("zk.connect", "http://XX.XX.XX.XX:2181";); // XX is the ip
> > props.put("serializer.class", "kafka.serializer.StringEncoder");
> > producer = new Producer(new ProducerConfig(props));
> >
> > and it fails on the fourth line with the following excetptions :
> >
> > kafka.common.FailedToSendMessageException: Failed to send messages after
> 3
> > tries
> > and
> > rg.I0Itec.zkclient.exception.ZkTimeoutException: Unable to connect to
> > zookeeper server within timeout: 400
> >
> >
> > Problems.
> > 1. The exception.
> > The bad parameters:
> > In the kafka quick start example I see that it needs only zk.connect,
> > serializer.class  . when I
> run
> > it it yells it needs metadata.broker.list in the constructor of the
> > Producer. Does it? So I feel the ip and port of the kafka server.
> > and btw - is it zk.connect or zookeeper connect?ZkTimeoutException:
> Unable
> > to connect to zookeeper server within timeout: 400
> >
> > 3. maven bad versions
> >
> > I go to the site, i see that the latest version is
>  kafka_2.8.0-0.8.0.
> > problem no 1 - I download it using intelij (I think it is maven
> > central) -
> > I get all related jars - only that the kafka jars are empty (contain
> > only manifest<
> >
> http://search.maven.org/#search%7Cgav%7C1%7Cg%3A%22org.apache.kafka%22%20AND%20a%3A%22kafka_2.8.0%22
> > >
> > ).
> >
> > problem no 2 - there are later versions then the one in the site. Are
> > they official.
> > any way, I downloaded org.apache.kafka:kafka_2.10:0.8.0
> > 4. Using wireshark I see three SYN & ACK (triple handshake) and than and
> > then FYN & ACK
> >  right after. in the logs of the zookeeper I see the following
> >   [2014-02-27 01:43:42,127] WARN EndOfStreamException: Unable to read
> > additional data from client sessionid 0x0, likely client has closed
> socket
> >
> >   (org.apache.zookeeper.server.NIOServerCnxn)
> >   Which means that I close the connection. Why?
> >
> > Oh, The horror. The horror. <
> > http://en.wikipedia.org/wiki/Heart_of_Darkness>
> >
>

Re: Producer fails when old brokers are replaced by new

[David Birdsong]

On Mon, Mar 3, 2014 at 4:00 PM, Guozhang Wang  wrote:

> Hi Chris,
>
> In 0.9 we will have just one "broker list", i.e. the list of brokers read
> from the config file will be updated during bootstraping and all the future
> metadata refresh operations. This feature should lift this limit you are
> describing, for example, if your broker list in config is {1,2,3}, and
> later on you extend the cluster to {1,2,3,4,5,6}, then now you can shut
> down 1,2,3 all at once.
>

But if you producer or consumer ever restarts and only knows about {1,2,3},
the problem still exists no?

This is why I bootstrap off of zk and expect to have to maintain an
accurate list of zk nodes to all processes.


>
> Guozhang
>
>
> On Mon, Mar 3, 2014 at 1:35 PM, Christofer Hedbrandh <
> christo...@knewton.com
> > wrote:
>
> > Thanks again Guozhang.
> >
> > There are still some details here that are unclear to me, but if what I
> am
> > describing is not a bug, do you think it is reasonable to file this as a
> > feature request? We agree that it is not ideal to have to keep "at least
> > one broker in the list is alive", when replacing a cluster i.e. migrating
> > from one set of brokers to another?
> >
> > Christofer
> >
> >
> >
> > On Wed, Feb 26, 2014 at 9:16 PM, Guozhang Wang 
> wrote:
> >
> > > kafka-preferred-replica-election.sh is only used to move leaders
> between
> > > brokers, as long as the broker in the broker.metadata.list, i.e. the
> > second
> > > broker list I mentioned in previous email is still alive then the
> > producer
> > > can learn the leader change from it.
> > >
> > > In terms of broker discovery, I think it depends on how you "define"
> the
> > > future. For example, originally there are 3 brokers 1,2,3, and you
> start
> > > the producer with metadata list = {1,2,3}, and later on another three
> > > brokers 4,5,6 are added, the producer can still find these newly added
> > > brokers. It is just that if all the brokers in the metadata list, i.e.
> > > 1,2,3 are gone, then the producer will not be able to refresh its
> > metadata.
> > >
> > > Guozhang
> > >
> > >
> > > On Wed, Feb 26, 2014 at 11:04 AM, Christofer Hedbrandh <
> > > christo...@knewton.com> wrote:
> > >
> > > > Thanks for your response Guozhang.
> > > >
> > > > I did make sure that new meta data is fetched before taking out the
> old
> > > > broker. I set the topic.metadata.refresh.interval.ms to something
> very
> > > > low,
> > > > and I confirm in the producer log that new meta data is actually
> > fetched,
> > > > after the new broker is brought up, and before the old broker is
> taken
> > > > down. Does this not mean that the dynamic current brokers list would
> > hold
> > > > the new broker at this point?
> > > >
> > > > If you are saying that the dynamic current brokers list is never used
> > for
> > > > fetching meta data, this does not explain how the producer does NOT
> > fail
> > > > when kafka-preferred-replica-election.sh makes the new broker become
> > the
> > > > leader.
> > > >
> > > > Lastly, if broker discovery is not a producer feature in 0.8.0
> Release,
> > > and
> > > > I have to "make sure at least one broker in the list is alive during
> > the
> > > > rolling bounce", is this a feature you are considering for future
> > > versions?
> > > >
> > > >
> > > >
> > > > On Wed, Feb 26, 2014 at 12:04 PM, Guozhang Wang 
> > > > wrote:
> > > >
> > > > > Hello Chris,
> > > > >
> > > > > The broker.metadata.list, once read in at start up time, will not
> be
> > > > > changed. In other words, during the life time of a producer it has
> > two
> > > > > lists of brokers:
> > > > >
> > > > > 1. The current brokers in the cluster that is returned in the
> > metadata
> > > > > request response, which is dynamic
> > > > >
> > > > > 2. The broker list that is used for bootstraping, this is read from
> > > > > broker.metadata.list and is fixed. This list could for example be a
> > VIP
> > > > and
> > > > > a hardware load balancer behind it will distribute the metadata
> > > requests
> > > > to
> > > > > the brokers.
> > > > >
> > > > > So in your case, the metadata list only has broker B, and once it
> is
> > > > taken
> > > > > out and the producer failed to send message to it and hence tries
> to
> > > > > refresh its metadata, it has no broker to go.
> > > > >
> > > > > Therefore, when you are trying to do a rolling bounce of the
> cluster
> > > to,
> > > > > for example, do a in-place upgrade, you need to make sure at least
> > one
> > > > > broker in the list is alive during the rolling bounce.
> > > > >
> > > > > Hope this helps.
> > > > >
> > > > > Guozhang
> > > > >
> > > > >
> > > > >
> > > > >
> > > > >
> > > > > On Wed, Feb 26, 2014 at 8:19 AM, Christofer Hedbrandh <
> > > > > christo...@knewton.com> wrote:
> > > > >
> > > > > > Hi all,
> > > > > >
> > > > > > I ran into a problem with the Kafka producer when attempting to
> > > replace
> > > > > all
> > > > > > the nodes in a 0.8.0 Beta1 Release Kafka cluster, with 0.8.0
> > Rel

Re: Producer fails when old brokers are replaced by new

[Guozhang Wang]

Hi Chris,

In 0.9 we will have just one "broker list", i.e. the list of brokers read
from the config file will be updated during bootstraping and all the future
metadata refresh operations. This feature should lift this limit you are
describing, for example, if your broker list in config is {1,2,3}, and
later on you extend the cluster to {1,2,3,4,5,6}, then now you can shut
down 1,2,3 all at once.

Guozhang


On Mon, Mar 3, 2014 at 1:35 PM, Christofer Hedbrandh  wrote:

> Thanks again Guozhang.
>
> There are still some details here that are unclear to me, but if what I am
> describing is not a bug, do you think it is reasonable to file this as a
> feature request? We agree that it is not ideal to have to keep "at least
> one broker in the list is alive", when replacing a cluster i.e. migrating
> from one set of brokers to another?
>
> Christofer
>
>
>
> On Wed, Feb 26, 2014 at 9:16 PM, Guozhang Wang  wrote:
>
> > kafka-preferred-replica-election.sh is only used to move leaders between
> > brokers, as long as the broker in the broker.metadata.list, i.e. the
> second
> > broker list I mentioned in previous email is still alive then the
> producer
> > can learn the leader change from it.
> >
> > In terms of broker discovery, I think it depends on how you "define" the
> > future. For example, originally there are 3 brokers 1,2,3, and you start
> > the producer with metadata list = {1,2,3}, and later on another three
> > brokers 4,5,6 are added, the producer can still find these newly added
> > brokers. It is just that if all the brokers in the metadata list, i.e.
> > 1,2,3 are gone, then the producer will not be able to refresh its
> metadata.
> >
> > Guozhang
> >
> >
> > On Wed, Feb 26, 2014 at 11:04 AM, Christofer Hedbrandh <
> > christo...@knewton.com> wrote:
> >
> > > Thanks for your response Guozhang.
> > >
> > > I did make sure that new meta data is fetched before taking out the old
> > > broker. I set the topic.metadata.refresh.interval.ms to something very
> > > low,
> > > and I confirm in the producer log that new meta data is actually
> fetched,
> > > after the new broker is brought up, and before the old broker is taken
> > > down. Does this not mean that the dynamic current brokers list would
> hold
> > > the new broker at this point?
> > >
> > > If you are saying that the dynamic current brokers list is never used
> for
> > > fetching meta data, this does not explain how the producer does NOT
> fail
> > > when kafka-preferred-replica-election.sh makes the new broker become
> the
> > > leader.
> > >
> > > Lastly, if broker discovery is not a producer feature in 0.8.0 Release,
> > and
> > > I have to "make sure at least one broker in the list is alive during
> the
> > > rolling bounce", is this a feature you are considering for future
> > versions?
> > >
> > >
> > >
> > > On Wed, Feb 26, 2014 at 12:04 PM, Guozhang Wang 
> > > wrote:
> > >
> > > > Hello Chris,
> > > >
> > > > The broker.metadata.list, once read in at start up time, will not be
> > > > changed. In other words, during the life time of a producer it has
> two
> > > > lists of brokers:
> > > >
> > > > 1. The current brokers in the cluster that is returned in the
> metadata
> > > > request response, which is dynamic
> > > >
> > > > 2. The broker list that is used for bootstraping, this is read from
> > > > broker.metadata.list and is fixed. This list could for example be a
> VIP
> > > and
> > > > a hardware load balancer behind it will distribute the metadata
> > requests
> > > to
> > > > the brokers.
> > > >
> > > > So in your case, the metadata list only has broker B, and once it is
> > > taken
> > > > out and the producer failed to send message to it and hence tries to
> > > > refresh its metadata, it has no broker to go.
> > > >
> > > > Therefore, when you are trying to do a rolling bounce of the cluster
> > to,
> > > > for example, do a in-place upgrade, you need to make sure at least
> one
> > > > broker in the list is alive during the rolling bounce.
> > > >
> > > > Hope this helps.
> > > >
> > > > Guozhang
> > > >
> > > >
> > > >
> > > >
> > > >
> > > > On Wed, Feb 26, 2014 at 8:19 AM, Christofer Hedbrandh <
> > > > christo...@knewton.com> wrote:
> > > >
> > > > > Hi all,
> > > > >
> > > > > I ran into a problem with the Kafka producer when attempting to
> > replace
> > > > all
> > > > > the nodes in a 0.8.0 Beta1 Release Kafka cluster, with 0.8.0
> Release
> > > > nodes.
> > > > > I started a producer/consumer test program to measure the clusters
> > > > > performance during the process, I added new brokers, I ran
> > > > > kafka-reassign-partitions.sh, and I removed the old brokers. When I
> > > > removed
> > > > > the old brokers my producer failed.
> > > > >
> > > > > The simplest scenario that I could come up with where I still see
> > this
> > > > > behavior is this. Using version 0.8.0 Release, we have a 1
> partition
> > > > topic
> > > > > with 2 replicas on 2 brokers, broker A and broker B. Broker A is
> > taken
> > > > 

Re: Producer fails when old brokers are replaced by new

[Christofer Hedbrandh]

Thanks again Guozhang.

There are still some details here that are unclear to me, but if what I am
describing is not a bug, do you think it is reasonable to file this as a
feature request? We agree that it is not ideal to have to keep "at least
one broker in the list is alive", when replacing a cluster i.e. migrating
from one set of brokers to another?

Christofer



On Wed, Feb 26, 2014 at 9:16 PM, Guozhang Wang  wrote:

> kafka-preferred-replica-election.sh is only used to move leaders between
> brokers, as long as the broker in the broker.metadata.list, i.e. the second
> broker list I mentioned in previous email is still alive then the producer
> can learn the leader change from it.
>
> In terms of broker discovery, I think it depends on how you "define" the
> future. For example, originally there are 3 brokers 1,2,3, and you start
> the producer with metadata list = {1,2,3}, and later on another three
> brokers 4,5,6 are added, the producer can still find these newly added
> brokers. It is just that if all the brokers in the metadata list, i.e.
> 1,2,3 are gone, then the producer will not be able to refresh its metadata.
>
> Guozhang
>
>
> On Wed, Feb 26, 2014 at 11:04 AM, Christofer Hedbrandh <
> christo...@knewton.com> wrote:
>
> > Thanks for your response Guozhang.
> >
> > I did make sure that new meta data is fetched before taking out the old
> > broker. I set the topic.metadata.refresh.interval.ms to something very
> > low,
> > and I confirm in the producer log that new meta data is actually fetched,
> > after the new broker is brought up, and before the old broker is taken
> > down. Does this not mean that the dynamic current brokers list would hold
> > the new broker at this point?
> >
> > If you are saying that the dynamic current brokers list is never used for
> > fetching meta data, this does not explain how the producer does NOT fail
> > when kafka-preferred-replica-election.sh makes the new broker become the
> > leader.
> >
> > Lastly, if broker discovery is not a producer feature in 0.8.0 Release,
> and
> > I have to "make sure at least one broker in the list is alive during the
> > rolling bounce", is this a feature you are considering for future
> versions?
> >
> >
> >
> > On Wed, Feb 26, 2014 at 12:04 PM, Guozhang Wang 
> > wrote:
> >
> > > Hello Chris,
> > >
> > > The broker.metadata.list, once read in at start up time, will not be
> > > changed. In other words, during the life time of a producer it has two
> > > lists of brokers:
> > >
> > > 1. The current brokers in the cluster that is returned in the metadata
> > > request response, which is dynamic
> > >
> > > 2. The broker list that is used for bootstraping, this is read from
> > > broker.metadata.list and is fixed. This list could for example be a VIP
> > and
> > > a hardware load balancer behind it will distribute the metadata
> requests
> > to
> > > the brokers.
> > >
> > > So in your case, the metadata list only has broker B, and once it is
> > taken
> > > out and the producer failed to send message to it and hence tries to
> > > refresh its metadata, it has no broker to go.
> > >
> > > Therefore, when you are trying to do a rolling bounce of the cluster
> to,
> > > for example, do a in-place upgrade, you need to make sure at least one
> > > broker in the list is alive during the rolling bounce.
> > >
> > > Hope this helps.
> > >
> > > Guozhang
> > >
> > >
> > >
> > >
> > >
> > > On Wed, Feb 26, 2014 at 8:19 AM, Christofer Hedbrandh <
> > > christo...@knewton.com> wrote:
> > >
> > > > Hi all,
> > > >
> > > > I ran into a problem with the Kafka producer when attempting to
> replace
> > > all
> > > > the nodes in a 0.8.0 Beta1 Release Kafka cluster, with 0.8.0 Release
> > > nodes.
> > > > I started a producer/consumer test program to measure the clusters
> > > > performance during the process, I added new brokers, I ran
> > > > kafka-reassign-partitions.sh, and I removed the old brokers. When I
> > > removed
> > > > the old brokers my producer failed.
> > > >
> > > > The simplest scenario that I could come up with where I still see
> this
> > > > behavior is this. Using version 0.8.0 Release, we have a 1 partition
> > > topic
> > > > with 2 replicas on 2 brokers, broker A and broker B. Broker A is
> taken
> > > > down. A producer is started with only broker B in the
> > > metadata.broker.list.
> > > > Broker A is brought back up. We let
> > > > topic.metadata.refresh.interval.msamount of time pass. Broker B is
> > > > taken down, and we get
> > > > kafka.common.FailedToSendMessageException after all the (many)
> retries
> > > have
> > > > failed.
> > > >
> > > > During my experimentation I have made sure that the producer fetches
> > meta
> > > > data before the old broker is taken down. And I have made sure that
> > > enough
> > > > retries with enough backoff time were used for the producer to not
> give
> > > up
> > > > prematurely.
> > > >
> > > > The documentation for the producer config metadata.broker.list
> suggests
> > 

Re: New Consumer API discussion

[Chris Riccomini]

Hey Guys,

Also, for reference, we'll be looking to implement new Samza consumers
which have these APIs:

http://samza.incubator.apache.org/learn/documentation/0.7.0/api/javadocs/or
g/apache/samza/system/SystemConsumer.html

http://samza.incubator.apache.org/learn/documentation/0.7.0/api/javadocs/or
g/apache/samza/checkpoint/CheckpointManager.html


Question (3) below is a result of having Samza's SystemConsumers poll
allow specific topic/partitions to be specified.

The split between consumer and checkpoint manager is the reason for
question (12) below.

Cheers,
Chris

On 3/3/14 10:19 AM, "Chris Riccomini"  wrote:

>Hey Guys,
>
>Sorry for the late follow up. Here are my questions/thoughts on the API:
>
>1. Why is the config String->Object instead of String->String?
>
>2. Are these Java docs correct?
>
>  KafkaConsumer(java.util.Map configs)
>  A consumer is instantiated by providing a set of key-value pairs as
>configuration and a ConsumerRebalanceCallback implementation
>
>There is no ConsumerRebalanceCallback parameter.
>
>3. Would like to have a method:
>
>  poll(long timeout, java.util.concurrent.TimeUnit timeUnit,
>TopicPartition... topicAndPartitionsToPoll)
>
>I see I can effectively do this by just fiddling with subscribe and
>unsubscribe before each poll. Is this a low-overhead operation? Can I just
>unsubscribe from everything after each poll, then re-subscribe to a topic
>the next iteration. I would probably be doing this in a fairly tight loop.
>
>4. The behavior of AUTO_OFFSET_RESET_CONFIG is overloaded. I think there
>are use cases for decoupling "what to do when no offset exists" from "what
>to do when I'm out of range". I might want to start from smallest the
>first time I run, but fail if I ever get offset out of range.
>
>5. ENABLE_JMX could use Java docs, even though it's fairly
>self-explanatory.
>
>6. Clarity about whether FETCH_BUFFER_CONFIG is per-topic/partition, or
>across all topic/partitions is useful. I believe it's per-topic/partition,
>right? That is, setting to 2 megs with two TopicAndPartitions would result
>in 4 megs worth of data coming in per fetch, right?
>
>7. What does the consumer do if METADATA_FETCH_TIMEOUT_CONFIG times out?
>Retry, or throw exception?
>
>8. Does RECONNECT_BACKOFF_MS_CONFIG apply to both metadata requests and
>fetch requests?
>
>9. What does SESSION_TIMEOUT_MS default to?
>
>10. Is this consumer thread-safe?
>
>11. How do you use a different offset management strategy? Your email
>implies that it's pluggable, but I don't see how. "The offset management
>strategy defaults to Kafka based offset management and the API provides a
>way for the user to use a customized offset store to manage the consumer's
>offsets."
>
>12. If I wish to decouple the consumer from the offset checkpointing, is
>it OK to use Joel's offset management stuff directly, rather than through
>the consumer's commit API?
>
>
>Cheers,
>Chris
>
>On 2/10/14 10:54 AM, "Neha Narkhede"  wrote:
>
>>As mentioned in previous emails, we are also working on a
>>re-implementation
>>of the consumer. I would like to use this email thread to discuss the
>>details of the public API. I would also like us to be picky about this
>>public api now so it is as good as possible and we don't need to break it
>>in the future.
>>
>>The best way to get a feel for the API is actually to take a look at the
>>javadoc>/
>>doc/kafka/clients/consumer/KafkaConsumer.html>,
>>the hope is to get the api docs good enough so that it is
>>self-explanatory.
>>You can also take a look at the configs
>>here>c
>>/kafka/clients/consumer/ConsumerConfig.html>
>>
>>Some background info on implementation:
>>
>>At a high level the primary difference in this consumer is that it
>>removes
>>the distinction between the "high-level" and "low-level" consumer. The
>>new
>>consumer API is non blocking and instead of returning a blocking
>>iterator,
>>the consumer provides a poll() API that returns a list of records. We
>>think
>>this is better compared to the blocking iterators since it effectively
>>decouples the threading strategy used for processing messages from the
>>consumer. It is worth noting that the consumer is entirely single
>>threaded
>>and runs in the user thread. The advantage is that it can be easily
>>rewritten in less multi-threading-friendly languages. The consumer
>>batches
>>data and multiplexes I/O over TCP connections to each of the brokers it
>>communicates with, for high throughput. The consumer also allows long
>>poll
>>to reduce the end-to-end message latency for low throughput data.
>>
>>The consumer provides a group management facility that supports the
>>concept
>>of a group with multiple consumer instances (just like the current
>>consumer). This is done through a custom heartbeat and group management
>>protocol transparent to the user. At the same time, it allows users th

Re: New Consumer API discussion

[Chris Riccomini]

Hey Guys,

Sorry for the late follow up. Here are my questions/thoughts on the API:

1. Why is the config String->Object instead of String->String?

2. Are these Java docs correct?

  KafkaConsumer(java.util.Map configs)
  A consumer is instantiated by providing a set of key-value pairs as
configuration and a ConsumerRebalanceCallback implementation

There is no ConsumerRebalanceCallback parameter.

3. Would like to have a method:

  poll(long timeout, java.util.concurrent.TimeUnit timeUnit,
TopicPartition... topicAndPartitionsToPoll)

I see I can effectively do this by just fiddling with subscribe and
unsubscribe before each poll. Is this a low-overhead operation? Can I just
unsubscribe from everything after each poll, then re-subscribe to a topic
the next iteration. I would probably be doing this in a fairly tight loop.

4. The behavior of AUTO_OFFSET_RESET_CONFIG is overloaded. I think there
are use cases for decoupling "what to do when no offset exists" from "what
to do when I'm out of range". I might want to start from smallest the
first time I run, but fail if I ever get offset out of range.

5. ENABLE_JMX could use Java docs, even though it's fairly
self-explanatory.

6. Clarity about whether FETCH_BUFFER_CONFIG is per-topic/partition, or
across all topic/partitions is useful. I believe it's per-topic/partition,
right? That is, setting to 2 megs with two TopicAndPartitions would result
in 4 megs worth of data coming in per fetch, right?

7. What does the consumer do if METADATA_FETCH_TIMEOUT_CONFIG times out?
Retry, or throw exception?

8. Does RECONNECT_BACKOFF_MS_CONFIG apply to both metadata requests and
fetch requests?

9. What does SESSION_TIMEOUT_MS default to?

10. Is this consumer thread-safe?

11. How do you use a different offset management strategy? Your email
implies that it's pluggable, but I don't see how. "The offset management
strategy defaults to Kafka based offset management and the API provides a
way for the user to use a customized offset store to manage the consumer's
offsets."

12. If I wish to decouple the consumer from the offset checkpointing, is
it OK to use Joel's offset management stuff directly, rather than through
the consumer's commit API?


Cheers,
Chris

On 2/10/14 10:54 AM, "Neha Narkhede"  wrote:

>As mentioned in previous emails, we are also working on a
>re-implementation
>of the consumer. I would like to use this email thread to discuss the
>details of the public API. I would also like us to be picky about this
>public api now so it is as good as possible and we don't need to break it
>in the future.
>
>The best way to get a feel for the API is actually to take a look at the
>javadocdoc/kafka/clients/consumer/KafkaConsumer.html>,
>the hope is to get the api docs good enough so that it is
>self-explanatory.
>You can also take a look at the configs
>here/kafka/clients/consumer/ConsumerConfig.html>
>
>Some background info on implementation:
>
>At a high level the primary difference in this consumer is that it removes
>the distinction between the "high-level" and "low-level" consumer. The new
>consumer API is non blocking and instead of returning a blocking iterator,
>the consumer provides a poll() API that returns a list of records. We
>think
>this is better compared to the blocking iterators since it effectively
>decouples the threading strategy used for processing messages from the
>consumer. It is worth noting that the consumer is entirely single threaded
>and runs in the user thread. The advantage is that it can be easily
>rewritten in less multi-threading-friendly languages. The consumer batches
>data and multiplexes I/O over TCP connections to each of the brokers it
>communicates with, for high throughput. The consumer also allows long poll
>to reduce the end-to-end message latency for low throughput data.
>
>The consumer provides a group management facility that supports the
>concept
>of a group with multiple consumer instances (just like the current
>consumer). This is done through a custom heartbeat and group management
>protocol transparent to the user. At the same time, it allows users the
>option to subscribe to a fixed set of partitions and not use group
>management at all. The offset management strategy defaults to Kafka based
>offset management and the API provides a way for the user to use a
>customized offset store to manage the consumer's offsets.
>
>A key difference in this consumer also is the fact that it does not depend
>on zookeeper at all.
>
>More details about the new consumer design are
>hereRewrite+Design>
>
>Please take a look at the new
>APIkafka/clients/consumer/KafkaConsumer.html>and
>give us any thoughts you may have.
>
>Thanks,
>Neha

Re: Consumer group ID for high level consumer

[Neha Narkhede]

Question  : How to make the consumer threads receive the older data that
was produced when the consumer threads aren't running ?

As long as you make sure the group.id doesn't change, the consumer will
always
consume data from where it left off during its last run. In other words,
the consumer
can consume data that was produced when it wasn't running, if it is always
started
using the same group.id

Thanks,
Neha


On Mon, Mar 3, 2014 at 8:02 AM, Binita Bharati wrote:

> Hi all,
>
> The HighLevelConsumer is working fine. Ref :
> https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example.
>
> I was expecting the consumer threads to be able to receive older data ( ie
> the data put into the topic when the consumer threads weren't running )
> That was the root of the problem, and once, I started my consumer
> application first, and then, generated fresh data into the topic, the
> consumer threads started receiving the new data.
>
> Question  : How to make the consumer threads receive the older data that
> was produced when the consumer threads aren't running ?
>
> Thanks
>
>
> On Wed, Feb 26, 2014 at 6:30 PM, Martin Kleppmann
> wrote:
>
> > Hi Binita,
> >
> > The consumer group (group.id) is a mechanism for sharing the load of
> > consuming a high-volume topic between multiple consumers. If you don't
> set
> > a group ID, each consumer consumes all the partitions of a topic. If you
> > set several consumers to the same group ID, the partitions of the topic
> > you're consuming will be shared amongst the consumers in that group, so
> > that each message is delivered to only one of the consumers in the group.
> >
> > You can set the group ID to be whatever you like.
> >
> > Hope that helps,
> > Martin
> >
> > On 25 Feb 2014, at 06:27, Binita Bharati 
> wrote:
> > > Hi all,
> > >
> > > I am referring to this e.g:
> > >
> https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
> > .
> > >
> > > What is the consumer group ID being referred here ?
> > >
> > > Thanks
> > > Binita
> >
> >
>

Re: to subscribe general user questions about Kafka

[Guozhang Wang]

Hi,

Please follow the instructions here:

http://kafka.apache.org/contact.html

Thanks,

Guozhang


On Mon, Mar 3, 2014 at 12:24 AM, 苏山  wrote:

> to subscribe general user questions about Kafka
>
> tks
>



-- 
-- Guozhang

Re: python brond - brod.base.OffsetOutOfRange: Code: 1

[Carlo Cabanilla]

It means that the offset you're requesting doesn't exist. Since you're
asking for offset 0, it's likely that that offset has already been expired
and you'll need to ask for a later offset. You can make an offsets request
using the EARLIEST_OFFSET constant to get the earliest offset that still
exists.


On Sat, Mar 1, 2014 at 5:09 AM, David Montgomery
wrote:

> How do I fix?
>
> In python
>
> for offset, message in brod.Kafka.fetch(kafka,pixel_topic, offset=0):
> print message
>
>
>  File
>
> "/usr/local/lib/python2.7/dist-packages/brod-0.3.2-py2.7.egg/brod/blocking.py",
> line 70, in _read
> return callback(output)
>   File
> "/usr/local/lib/python2.7/dist-packages/brod-0.3.2-py2.7.egg/brod/base.py",
> line 588, in _read_response
> raise error_codes.get(error_code, UnknownError)('Code:
> {0}'.format(error_code))
> brod.base.OffsetOutOfRange: Code: 1
>
>
> In kafka logs...
>
>
> [2014-03-01 09:58:30,886] INFO Closing socket connection to /
> 222.127.178.107.
> (kafka.network.Processor)
> [2014-03-01 10:01:29,518] INFO Closing socket connection to /
> 222.127.178.107.
> (kafka.network.Processor)
> [2014-03-01 10:01:41,473] ERROR error when processing request
> FetchRequest(topic:topic-pixel, part:0 offset:0 maxSize:1048576)
> (kafka.server.KafkaRequestHandlers)
> kafka.common.OffsetOutOfRangeException: offset 0 is out of range
> at kafka.log.Log$.findRange(Log.scala:46)
> at kafka.log.Log.read(Log.scala:264)
> at
>
> kafka.server.KafkaRequestHandlers.kafka$server$KafkaRequestHandlers$$readMessageSet(KafkaRequestHandlers.scala:112)
> at
>
> kafka.server.KafkaRequestHandlers.handleFetchRequest(KafkaRequestHandlers.scala:92)
> at
>
> kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$2.apply(KafkaRequestHandlers.scala:39)
> at
>
> kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$2.apply(KafkaRequestHandlers.scala:39)
> at kafka.network.Processor.handle(SocketServer.scala:296)
> at kafka.network.Processor.read(SocketServer.scala:319)
> at kafka.network.Processor.run(SocketServer.scala:214)
> at java.lang.Thread.run(Thread.java:744)
> [2014-03-01 10:01:41,573] INFO Closing socket connection to /
> 222.127.178.107.
> (kafka.network.Processor)
>

Re: Consumer group ID for high level consumer

[Binita Bharati]

Hi all,

The HighLevelConsumer is working fine. Ref :
https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example.

I was expecting the consumer threads to be able to receive older data ( ie
the data put into the topic when the consumer threads weren't running )
That was the root of the problem, and once, I started my consumer
application first, and then, generated fresh data into the topic, the
consumer threads started receiving the new data.

Question  : How to make the consumer threads receive the older data that
was produced when the consumer threads aren't running ?

Thanks


On Wed, Feb 26, 2014 at 6:30 PM, Martin Kleppmann
wrote:

> Hi Binita,
>
> The consumer group (group.id) is a mechanism for sharing the load of
> consuming a high-volume topic between multiple consumers. If you don't set
> a group ID, each consumer consumes all the partitions of a topic. If you
> set several consumers to the same group ID, the partitions of the topic
> you're consuming will be shared amongst the consumers in that group, so
> that each message is delivered to only one of the consumers in the group.
>
> You can set the group ID to be whatever you like.
>
> Hope that helps,
> Martin
>
> On 25 Feb 2014, at 06:27, Binita Bharati  wrote:
> > Hi all,
> >
> > I am referring to this e.g:
> > https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example
> .
> >
> > What is the consumer group ID being referred here ?
> >
> > Thanks
> > Binita
>
>

Re: How ot had scala libraries to path

[Jun Rao]

Hmm, not sure how duid picks up the jars. Perhaps you can ask duid mailing
list? Do you know if duid is on Kafka 0.7 or 0.8?

Thanks,

Jun


On Mon, Mar 3, 2014 at 3:37 AM, David Montgomery
wrote:

> I have reverted back to 7.  I can produce and consume using the python lib
> brod.
>
>
>
>
>
>
> Druid consumer gives this error.
>
> My libs are here.  So..kafka cant find scala?
>
>
> /var/lib/kafka-0.7.2-incubating-src/core/lib_managed/scala_2.8.0/test/scalatest-1.2.jar
>
> /var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.7.7/lib/scala-compiler.jar
>
> /var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.7.7/lib/scala-library.jar
>
> /var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.8.0/lib/scala-compiler.jar
>
> /var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.8.0/lib/scala-library.jar
>
>
> Here is how I install:]'cd kafka-0.7.2-incubating-src
> ./sbt update
> ./sbt package
> echo 'KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true
> -Dcom.sun.management.jmxremote.authenticate=false
> -Dcom.sun.management.jmxremote.ssl=false"' | tee -a
> /var/lib/kafka-0.7.2-incubating-src/bin/kafka-run-class.sh
> echo 'export JMX_PORT=${JMX_PORT:-}' | tee -a
> /var/lib/kafka-0.7.2-incubating-src/bin/kafka-server-start.sh
>
> Here is how I start
> /var/lib/kafka-0.7.2-incubating-src/bin/kafka-server-start.sh
> /var/lib/kafka-0.7.2-incubating-src/config/server.properties
>
>
>
> ing SASL (unknown error)
> 2014-03-03 11:19:40,237 INFO [main]
> com.metamx.common.lifecycle.Lifecycle$AnnotationBasedHandler - Invoking
> start method[public void io.druid.segment.realtime.RealtimeManager.start()
> throws java.io.IOException] on
> object[io.druid.segment.realtime.RealtimeManager@73ed27ed].
> 2014-03-03 11:19:40,241 INFO [main]
> io.druid.segment.realtime.RealtimeManager - Calling the FireDepartment and
> getting a Firehose.
> 2014-03-03 11:19:40,249 ERROR [main] io.druid.cli.CliBroker - Error when
> starting up.  Failing.
> java.lang.reflect.InvocationTargetException
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
>
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> at
>
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:606)
> at
>
> com.metamx.common.lifecycle.Lifecycle$AnnotationBasedHandler.start(Lifecycle.java:331)
> at com.metamx.common.lifecycle.Lifecycle.start(Lifecycle.java:250)
> at io.druid.guice.LifecycleModule$2.start(LifecycleModule.java:136)
> at io.druid.cli.GuiceRunnable.initLifecycle(GuiceRunnable.java:72)
> at io.druid.cli.ServerRunnable.run(ServerRunnable.java:40)
> at io.druid.cli.Main.main(Main.java:91)
> Caused by: java.lang.NoClassDefFoundError: scala/ScalaObject
> at java.lang.ClassLoader.defineClass1(Native Method)
> at java.lang.ClassLoader.defineClass(ClassLoader.java:800)
> at
> java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
> at java.net.URLClassLoader.defineClass(URLClassLoader.java:449)
> at java.net.URLClassLoader.access$100(URLClassLoader.java:71)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> at
>
> io.druid.firehose.kafka.KafkaSevenFirehoseFactory.connect(KafkaSevenFirehoseFactory.java:89)
> at
> io.druid.segment.realtime.FireDepartment.connect(FireDepartment.java:90)
> at
>
> io.druid.segment.realtime.RealtimeManager$FireChief.init(RealtimeManager.java:150)
> at
> io.druid.segment.realtime.RealtimeManager.start(RealtimeManager.java:87)
> ... 10 more
> Caused by: java.lang.ClassNotFoundException: scala.ScalaObject
> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> ... 25 more
>
>
>
> On Wed, Feb 26, 2014 at 12:40 PM, David Montgomery <
> davidmontgom...@gmail.com> wrote:
>
> > Hi,
> >
> > This is how I start kafka.
> >
> > command = /var/lib/kafka-<%=@version%>-src/bin/kafka-server-start.sh
> > /var/lib/kafka-<%=@version%>-src/config/server.properties
> >
> > In another application I get teh below error.  The suggestion is to add
> > the scalar libraries to the path.  How do I do that?
> >
> >
> > thanks
> >
> >
> > java.lang.reflect.
> > InvocationTargetException
> > at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> > at
> >

to subscribe general user questions about Kafka

[苏山]

to subscribe general user questions about Kafka

tks

Re: How ot had scala libraries to path

[David Montgomery]

I have reverted back to 7.  I can produce and consume using the python lib
brod.






Druid consumer gives this error.

My libs are here.  So..kafka cant find scala?

/var/lib/kafka-0.7.2-incubating-src/core/lib_managed/scala_2.8.0/test/scalatest-1.2.jar
/var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.7.7/lib/scala-compiler.jar
/var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.7.7/lib/scala-library.jar
/var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.8.0/lib/scala-compiler.jar
/var/lib/kafka-0.7.2-incubating-src/project/boot/scala-2.8.0/lib/scala-library.jar


Here is how I install:]'cd kafka-0.7.2-incubating-src
./sbt update
./sbt package
echo 'KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true
-Dcom.sun.management.jmxremote.authenticate=false
-Dcom.sun.management.jmxremote.ssl=false"' | tee -a
/var/lib/kafka-0.7.2-incubating-src/bin/kafka-run-class.sh
echo 'export JMX_PORT=${JMX_PORT:-}' | tee -a
/var/lib/kafka-0.7.2-incubating-src/bin/kafka-server-start.sh

Here is how I start
/var/lib/kafka-0.7.2-incubating-src/bin/kafka-server-start.sh
/var/lib/kafka-0.7.2-incubating-src/config/server.properties



ing SASL (unknown error)
2014-03-03 11:19:40,237 INFO [main]
com.metamx.common.lifecycle.Lifecycle$AnnotationBasedHandler - Invoking
start method[public void io.druid.segment.realtime.RealtimeManager.start()
throws java.io.IOException] on
object[io.druid.segment.realtime.RealtimeManager@73ed27ed].
2014-03-03 11:19:40,241 INFO [main]
io.druid.segment.realtime.RealtimeManager - Calling the FireDepartment and
getting a Firehose.
2014-03-03 11:19:40,249 ERROR [main] io.druid.cli.CliBroker - Error when
starting up.  Failing.
java.lang.reflect.InvocationTargetException
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at
com.metamx.common.lifecycle.Lifecycle$AnnotationBasedHandler.start(Lifecycle.java:331)
at com.metamx.common.lifecycle.Lifecycle.start(Lifecycle.java:250)
at io.druid.guice.LifecycleModule$2.start(LifecycleModule.java:136)
at io.druid.cli.GuiceRunnable.initLifecycle(GuiceRunnable.java:72)
at io.druid.cli.ServerRunnable.run(ServerRunnable.java:40)
at io.druid.cli.Main.main(Main.java:91)
Caused by: java.lang.NoClassDefFoundError: scala/ScalaObject
at java.lang.ClassLoader.defineClass1(Native Method)
at java.lang.ClassLoader.defineClass(ClassLoader.java:800)
at
java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
at java.net.URLClassLoader.defineClass(URLClassLoader.java:449)
at java.net.URLClassLoader.access$100(URLClassLoader.java:71)
at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
at
io.druid.firehose.kafka.KafkaSevenFirehoseFactory.connect(KafkaSevenFirehoseFactory.java:89)
at
io.druid.segment.realtime.FireDepartment.connect(FireDepartment.java:90)
at
io.druid.segment.realtime.RealtimeManager$FireChief.init(RealtimeManager.java:150)
at
io.druid.segment.realtime.RealtimeManager.start(RealtimeManager.java:87)
... 10 more
Caused by: java.lang.ClassNotFoundException: scala.ScalaObject
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
... 25 more



On Wed, Feb 26, 2014 at 12:40 PM, David Montgomery <
davidmontgom...@gmail.com> wrote:

> Hi,
>
> This is how I start kafka.
>
> command = /var/lib/kafka-<%=@version%>-src/bin/kafka-server-start.sh
> /var/lib/kafka-<%=@version%>-src/config/server.properties
>
> In another application I get teh below error.  The suggestion is to add
> the scalar libraries to the path.  How do I do that?
>
>
> thanks
>
>
> java.lang.reflect.
> InvocationTargetException
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:606)
> at
> com.metamx.common.lifecycle.Lifecycle$AnnotationBasedHandler.start(Lifecycle.java:331)
> at com.metamx.common.lifecycle.Lifecycle.start(Lifecycle.java:250)
> at io.druid.guice.L