kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jimmy John <jj...@livefyre.com>
Subject Re: OffsetOutOfRange errors
Date Fri, 07 Nov 2014 18:30:22 GMT
The current setting is to commit to ZK every 100 messages read.

The read buffer size is 262144 bytes. So we will read in a bunch of
messages in a batch. And while iterating through those messages, we commit
the offset to ZK every 100.

jim

On Fri, Nov 7, 2014 at 10:13 AM, Guozhang Wang <wangguoz@gmail.com> wrote:

> Hi Jim,
>
> When messages gets cleaned based on data retention policy (by time or by
> size), the brokers will not inform ZK for the deletion event. The
> underlying assumption is that when consumers are fetching data at around
> the tail of the log (i.e. they are not much lagging, which is normal cases)
> they should be continuously update the consumed offsets in ZK and hence
> that offsets will be valid most of the time. When consumers are lagging
> behind and the old messages are cleaned they will get this exception, and
> consumers need to handle it by resetting their offset to, e.g. the head of
> the log.
>
> How frequent do your clients read / write the offsets in ZK?
>
> Guozhang
>
> On Thu, Nov 6, 2014 at 6:23 PM, Jimmy John <jjohn@livefyre.com> wrote:
>
> > Hello,
> >
> >   I understand what this error means, just not sure why I keep running
> into
> > it after 24-48 hrs of running fine consuming > 300 messages / second.
> >
> >   What happens when a kafka log rolls over and some old records are aged
> > out? I mean what happens to the offsets? We are using a python client
> which
> > stores the offsets in ZK. But in the middle of the run, say after 2 days
> or
> > so, suddenly it gets this error.
> >
> > The only possibility is that the older records have aged off and ZK still
> > has the offset which is no longer applicable...How does the java client
> > deal with this? Does kafka inform ZK that records have been aged off and
> > update the offset or something?
> >
> > Here is the error i see in the broker logs
> >
> > [2014-11-07 01:40:32,478] ERROR [KafkaApi-11] Error when processing fetch
> > request for partition [activity.stream,3] offset 8013827 from consumer
> > with
> > correlation id 73 (kafka.server.KafkaApis)
> >
> >  kafka.common.OffsetOutOfRangeException: Request for offset 8013827 but
> we
> > only have log segments in the range 8603331 to 11279773.
> >
> >      at kafka.log.Log.read(Log.scala:380)
> >
> >      at
> >
> >
> kafka.server.KafkaApis.kafka$server$KafkaApis$$readMessageSet(KafkaApis.scala:530)
> >
> >      at
> >
> >
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$readMessageSets$1.apply(KafkaApis.scala:476)
> >
> >      at
> >
> >
> kafka.server.KafkaApis$$anonfun$kafka$server$KafkaApis$$readMessageSets$1.apply(KafkaApis.scala:471)
> >
> >      at
> >
> >
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
> >
> >      at
> >
> >
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:233)
> >
> >      at scala.collection.immutable.Map$Map3.foreach(Map.scala:164)
> >
> >     at
> > scala.collection.TraversableLike$class.map(TraversableLike.scala:233)
> >
> >      at scala.collection.immutable.Map$Map3.map(Map.scala:144)
> >
> >      at
> >
> >
> kafka.server.KafkaApis.kafka$server$KafkaApis$$readMessageSets(KafkaApis.scala:471)
> >
> >      at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:437)
> >
> >      at kafka.server.KafkaApis.handle(KafkaApis.scala:186)
> >
> >      at
> kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:42)
> >
> >      at java.lang.Thread.run(Thread.java:745)
> >
> >
> > thx
> >
> > Jim
> >
>
>
>
> --
> -- Guozhang
>

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message