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 > > > > > >
