Jun, we've been using Kafka for more than two years. Both the key and the value are type string. That doesn't seem to be the problem.
I just can't start the application when setting the callback handler which I tried with string and byte[]. The reason I want to use the handler is to send metrics that statsd. What else can I try to make this callback work? Sent from my Nexus 4. On Jan 30, 2014 8:02 AM, "Jun Rao" <[email protected]> wrote: > Yes, you should implement kafka.javaapi.producer.async package. Internally, > we wrap that callback with a scala callback. When instantiating the > producer, you need to provide 2 types, the first one for key and the second > one for value. Make sure the second one is of type byte[]. > > Thanks, > > Jun > > > On Wed, Jan 29, 2014 at 10:17 PM, Patricio Echagüe <[email protected] > >wrote: > > > It's String. I also tried with the generic type String. > > > > The CallbackHandler interface I implement is the one in > > kafka.javaapi.producer.async package. Is that the right one? > > I'm a bit confused because the exception mentions kafka.producer.async. > > CallbackHandler. > > > > > > On Wed, Jan 29, 2014 at 9:05 PM, Jun Rao <[email protected]> wrote: > > > > > Is your producer instantiated with type byte[]? > > > > > > Thanks, > > > > > > Jun > > > > > > > > > On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <[email protected] > > > >wrote: > > > > > > > I'm trying to set a callback handler from java. > > > > > > > > For that, I created a Callback Handler this way: > > > > > > > > public class KafkaAsyncCallbackHandler implements > > > CallbackHandler<byte[]> { > > > > } > > > > > > > > and set the property > > > > > > > > callback.handler=com.lucid.kafka.KafkaAsyncCallbackHandler > > > > > > > > > > > > But on runtime I get this exception coming from kafka: > > > > > > > > Caused by: java.lang.ClassCastException: > > > > com.lucid.kafka.KafkaAsyncCallbackHandler cannot be cast to > > > > kafka.producer.async.CallbackHandler > > > > at kafka.producer.ProducerPool.<init>(ProducerPool.scala:62) > > > > at kafka.javaapi.producer.Producer.<init>(Producer.scala:41) > > > > at > > > > > > > > > > com.lucid.dao.queue.impl.kafka.KafkaProducer.<init>(KafkaProducer.java:29) > > > > at com.lucid.dao.guice.DAOModule.provideProducer(DAOModule.java:448) > > > > > > > > We are on kafka 0.7.1 > > > > > > > > Am I doing anything wrong here? > > > > > > > > Thanks > > > > Patricio > > > > > > > > > >
