Handling un-decodable messages (Kafka 0.8.0)

[Christofer Hedbrandh <ch...@knewton.com>]

I have a custom Decoder for my messages (Thrift). I want to be able to
handle "bad" messages that I can't decode. When the ConsumerIterator
encounters a bad message, the exception thrown by my Decoder bubbles up and
I can catch it and handle it. Subsequent calls to the ConsumerIterator give
me IllegalStateException: "Iterator is in failed state" however.

What is the preferred way to handle messages that my custom Decoder can't
decode, and then continue consuming messages?

It doesn't sound reasonable to me that one bad message causes the whole
consumer to fall over.


Kafka version: 0.8.0

Stacktrace:
java.lang.IllegalStateException: Iterator is in failed state
    at kafka.utils.IteratorTemplate.hasNext(Unknown Source)
~[kafka_2.10-0.8.0.jar:0.8.0]

Re: Handling un-decodable messages (Kafka 0.8.0)

[Tom Brown <to...@gmail.com>]

Another way would be to have your custom decoder return an object that can
be recognized as an error.

We have a decoder that splits binary data into a series of records. If any
part of the binary data is corrupt, the decoder can be configured to either
throw an exception or add an "error record" to the output series.

--Tom

On Monday, June 9, 2014, Jun Rao <ju...@gmail.com> wrote:

> This is handled better in 0.8.1.1. In the same situation, you can resume
> the iteration on the next message. To do this in 0.8.0, you will have to
> get the binary data in ConsumerIterator and run the decoder in the consumer
> code.
>
> Thanks,
>
> Jun
>
>
> On Mon, Jun 9, 2014 at 9:40 AM, Christofer Hedbrandh <
> christofer@knewton.com <javascript:;>
> > wrote:
>
> > I have a custom Decoder for my messages (Thrift). I want to be able to
> > handle "bad" messages that I can't decode. When the ConsumerIterator
> > encounters a bad message, the exception thrown by my Decoder bubbles up
> and
> > I can catch it and handle it. Subsequent calls to the ConsumerIterator
> give
> > me IllegalStateException: "Iterator is in failed state" however.
> >
> > What is the preferred way to handle messages that my custom Decoder can't
> > decode, and then continue consuming messages?
> >
> > It doesn't sound reasonable to me that one bad message causes the whole
> > consumer to fall over.
> >
> >
> > Kafka version: 0.8.0
> >
> > Stacktrace:
> > java.lang.IllegalStateException: Iterator is in failed state
> >     at kafka.utils.IteratorTemplate.hasNext(Unknown Source)
> > ~[kafka_2.10-0.8.0.jar:0.8.0]
> >
>

Re: Handling un-decodable messages (Kafka 0.8.0)

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

This is handled better in 0.8.1.1. In the same situation, you can resume
the iteration on the next message. To do this in 0.8.0, you will have to
get the binary data in ConsumerIterator and run the decoder in the consumer
code.

Thanks,

Jun


On Mon, Jun 9, 2014 at 9:40 AM, Christofer Hedbrandh <christofer@knewton.com
> wrote:

> I have a custom Decoder for my messages (Thrift). I want to be able to
> handle "bad" messages that I can't decode. When the ConsumerIterator
> encounters a bad message, the exception thrown by my Decoder bubbles up and
> I can catch it and handle it. Subsequent calls to the ConsumerIterator give
> me IllegalStateException: "Iterator is in failed state" however.
>
> What is the preferred way to handle messages that my custom Decoder can't
> decode, and then continue consuming messages?
>
> It doesn't sound reasonable to me that one bad message causes the whole
> consumer to fall over.
>
>
> Kafka version: 0.8.0
>
> Stacktrace:
> java.lang.IllegalStateException: Iterator is in failed state
>     at kafka.utils.IteratorTemplate.hasNext(Unknown Source)
> ~[kafka_2.10-0.8.0.jar:0.8.0]
>