But we want a programmatic way to skip in this error case, can the high
level consumer provide a flag to skip corrupted message in its iterator?


On Tue, Aug 4, 2015 at 10:43 PM, Gwen Shapira <g...@confluent.io> wrote:

> The high level consumer stores its state in ZooKeeper. Theoretically, you
> should be able to go into ZooKeeper, find the consumer-group, topic and
> partition, and increment the offset past the "corrupt" point.
>
> On Tue, Aug 4, 2015 at 10:23 PM, Henry Cai <h...@pinterest.com.invalid>
> wrote:
>
> > Hi,
> >
> > We are using the Kafka high-level consumer 8.1.1, somehow we got a
> > corrupted message in the topic.  We are not sure the root cause of this,
> > but the problem we are having now is the HL consumer is stuck in that
> > position:
> >
> > kafka.message.InvalidMessageException: Message is corrupt (stored crc =
> > 537685622, computed crc = 36513351)
> >
> >         at kafka.message.Message.ensureValid(Message.scala:166)
> >
> >         at
> > kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:102)
> >
> >         at
> > kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:33)
> >
> >         at
> > kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:66)
> >
> >         at
> kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:58)
> >
> >
> > If we try to ignore that exception and iterate to the next message, the
> > iterator couldn't pass that error state:
> >
> > java.lang.IllegalStateException: Iterator is in failed state
> >
> >         at
> kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54)
> >
> >
> > By looking at the code, looks like you can only calling
> > IteratorTemplate.resetState() to clear the state, but this is an internal
> > method, is this the right way to workaround this problem?
> >
>

Reply via email to