Cannot cast to kafka.producer.async.CallbackHandler

[Patricio Echagüe <pa...@gmail.com>]

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

Re: Cannot cast to kafka.producer.async.CallbackHandler

[Neha Narkhede <ne...@gmail.com>]

Sure. However, we are working actively on a new producer API for Kafka. It
will be good if you can take a look at it and provide feedback -
https://www.mail-archive.com/dev@kafka.apache.org/msg07187.html

Thanks,
Neha


On Thu, Jan 30, 2014 at 2:28 PM, Patricio Echagüe <pa...@gmail.com>wrote:

> Update:
>
> There was no way to make it work with the javaapi one.
>
> I made it work using the scala interface from java. The code looks a bit
> ugly as I had to create a scala set from java.
>
> For example:
>
> @Override
> public Seq<QueueItem<String>> afterDequeuingExistingData(QueueItem<String>
> queueItem) {
>  if (queueItem == null) {
> return new scala.collection.mutable.ArraySeq<QueueItem<String>>(0);
>  }
>
> MutableList<QueueItem<String>> seq = new MutableList<QueueItem<String>>();
>  seq.$plus$eq(queueItem);
> return seq;
> }
>
> Should I file an issue for this?
>
>
> On Thu, Jan 30, 2014 at 9:53 AM, Patricio Echagüe <patricioe@gmail.com
> >wrote:
>
> > one more thing. Using the scala callback handler from java code seems to
> > work but I'm having a hard time creating scala Seqs from java code to
> make
> > my handler compatible with the scala signature.
> >
> >
> > On Thu, Jan 30, 2014 at 8:10 AM, Patricio Echagüe <patricioe@gmail.com
> >wrote:
> >
> >> 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" <ju...@gmail.com> 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 <
> patricioe@gmail.com
> >>> >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 <ju...@gmail.com> wrote:
> >>> >
> >>> > > Is your producer instantiated with type byte[]?
> >>> > >
> >>> > > Thanks,
> >>> > >
> >>> > > Jun
> >>> > >
> >>> > >
> >>> > > On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <
> >>> patricioe@gmail.com
> >>> > > >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
> >>> > > >
> >>> > >
> >>> >
> >>>
> >>
> >
>

Re: Cannot cast to kafka.producer.async.CallbackHandler

[Patricio Echagüe <pa...@gmail.com>]

Update:

There was no way to make it work with the javaapi one.

I made it work using the scala interface from java. The code looks a bit
ugly as I had to create a scala set from java.

For example:

@Override
public Seq<QueueItem<String>> afterDequeuingExistingData(QueueItem<String>
queueItem) {
 if (queueItem == null) {
return new scala.collection.mutable.ArraySeq<QueueItem<String>>(0);
 }

MutableList<QueueItem<String>> seq = new MutableList<QueueItem<String>>();
 seq.$plus$eq(queueItem);
return seq;
}

Should I file an issue for this?


On Thu, Jan 30, 2014 at 9:53 AM, Patricio Echagüe <pa...@gmail.com>wrote:

> one more thing. Using the scala callback handler from java code seems to
> work but I'm having a hard time creating scala Seqs from java code to make
> my handler compatible with the scala signature.
>
>
> On Thu, Jan 30, 2014 at 8:10 AM, Patricio Echagüe <pa...@gmail.com>wrote:
>
>> 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" <ju...@gmail.com> 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 <patricioe@gmail.com
>>> >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 <ju...@gmail.com> wrote:
>>> >
>>> > > Is your producer instantiated with type byte[]?
>>> > >
>>> > > Thanks,
>>> > >
>>> > > Jun
>>> > >
>>> > >
>>> > > On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <
>>> patricioe@gmail.com
>>> > > >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
>>> > > >
>>> > >
>>> >
>>>
>>
>

Re: Cannot cast to kafka.producer.async.CallbackHandler

[Patricio Echagüe <pa...@gmail.com>]

one more thing. Using the scala callback handler from java code seems to
work but I'm having a hard time creating scala Seqs from java code to make
my handler compatible with the scala signature.


On Thu, Jan 30, 2014 at 8:10 AM, Patricio Echagüe <pa...@gmail.com>wrote:

> 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" <ju...@gmail.com> 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 <patricioe@gmail.com
>> >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 <ju...@gmail.com> wrote:
>> >
>> > > Is your producer instantiated with type byte[]?
>> > >
>> > > Thanks,
>> > >
>> > > Jun
>> > >
>> > >
>> > > On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <
>> patricioe@gmail.com
>> > > >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
>> > > >
>> > >
>> >
>>
>

Re: Cannot cast to kafka.producer.async.CallbackHandler

[Patricio Echagüe <pa...@gmail.com>]

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" <ju...@gmail.com> 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 <patricioe@gmail.com
> >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 <ju...@gmail.com> wrote:
> >
> > > Is your producer instantiated with type byte[]?
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > >
> > > On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <patricioe@gmail.com
> > > >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
> > > >
> > >
> >
>

Re: Cannot cast to kafka.producer.async.CallbackHandler

[Jun Rao <ju...@gmail.com>]

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 <pa...@gmail.com>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 <ju...@gmail.com> wrote:
>
> > Is your producer instantiated with type byte[]?
> >
> > Thanks,
> >
> > Jun
> >
> >
> > On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <patricioe@gmail.com
> > >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
> > >
> >
>

Re: Cannot cast to kafka.producer.async.CallbackHandler

[Patricio Echagüe <pa...@gmail.com>]

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 <ju...@gmail.com> wrote:

> Is your producer instantiated with type byte[]?
>
> Thanks,
>
> Jun
>
>
> On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <patricioe@gmail.com
> >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
> >
>

Re: Cannot cast to kafka.producer.async.CallbackHandler

[Jun Rao <ju...@gmail.com>]

Is your producer instantiated with type byte[]?

Thanks,

Jun


On Wed, Jan 29, 2014 at 7:25 PM, Patricio Echagüe <pa...@gmail.com>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
>