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