The kafka logs have this
[2015-05-19 17:13:39,772] ERROR [KafkaApi-0] Error when processing fetch
request for partition [warehouse_prices,0] offset 73792051 from consumer with
correlation id 0 (kafka.server.KafkaApis)
kafka.common.OffsetOutOfRangeException: Request for offset 73792051 but we only
have log segments in the range 74630863 to 75835831.
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:206)
How can that be? How can the requestor be behind kafka?
> On 19 May 2015, at 14:54, Benjamin Cuthbert <[email protected]> wrote:
>
> Hi Jeff,
>
> So I looked at the docs and I reset the following property:
>
> SpoutConfig spoutConfig = new SpoutConfig(
> hosts,
> topic, // topic to read from
> KAFKA_STORM_DIR, // the root path in Zookeeper
> for the spout to store the consumer offsets
> newSpoutId); // an id for this consumer for
> storing the consumer offsets in Zookeeper
>
> //Check if we should be consuming messages from the beginning
> spoutConfig.forceFromStart = consumeFromBeginning;
> spoutConfig.maxOffsetBehind = Long.MAX_VALUE;
> spoutConfig.useStartOffsetTimeIfOffsetOutOfRange = false;
>
> But after an hour of processing details I see
>
> 2015-05-19T13:13:03.242+0000 s.k.KafkaUtils [ERROR] Error fetching data from
> [Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
> partition=0}] for topic [warehouse_prices]: [OFFSET_OUT_OF_RANGE]
> 2015-05-19T13:13:03.242+0000 s.k.KafkaSpout [WARN] Fetch failed
> storm.kafka.FailedFetchException: Error fetching data from
> [Partition{host=price-engine-demo-server.c.celertech-01.internal:9092,
> partition=0}] for topic [warehouse_prices]: [OFFSET_OUT_OF_RANGE]
> at storm.kafka.KafkaUtils.fetchMessages(KafkaUtils.java:190)
> ~[celertech-analytics-dependencies-DEVELOP-HEAD-SNAPSHOT.jar:na]
> at storm.kafka.PartitionManager.fill(PartitionManager.java:162)
> ~[celertech-analytics-dependencies-DEVELOP-HEAD-SNAPSHOT.jar:na]
> at storm.kafka.PartitionManager.next(PartitionManager.java:124)
> ~[celertech-analytics-dependencies-DEVELOP-HEAD-SNAPSHOT.jar:na]
> at storm.kafka.KafkaSpout.nextTuple(KafkaSpout.java:141)
> ~[celertech-analytics-dependencies-DEVELOP-HEAD-SNAPSHOT.jar:na]
> at
> backtype.storm.daemon.executor$fn__4654$fn__4669$fn__4698.invoke(executor.clj:565)
> [storm-core-0.9.4.jar:0.9.4]
> at backtype.storm.util$async_loop$fn__458.invoke(util.clj:463)
> [storm-core-0.9.4.jar:0.9.4]
> at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
> at java.lang.Thread.run(Thread.java:745) [na:1.7.0_79]
>
>
>> On 18 May 2015, at 22:13, Benjamin Cuthbert <[email protected]
>> <mailto:[email protected]>> wrote:
>>
>> Thanks Jeff,
>>
>> So I looked over the docs but I don’t understand is it runs for 2+ hours
>> then just starts going:
>>
>> 2015-05-18T22:12:53.673+0100 s.k.PartitionManager [WARN] Using new offset:
>> 64429892
>> 2015-05-18T22:12:53.705+0100 s.k.KafkaUtils [WARN] Got fetch request with
>> offset out of range: [63610973]; retrying with default start offset time
>> from configuration. configured start offset time: [-2]
>> 2015-05-18T22:12:53.743+0100 s.k.PartitionManager [WARN] Using new offset:
>> 64429934
>> 2015-05-18T22:12:53.773+0100 s.k.KafkaUtils [WARN] Got fetch request with
>> offset out of range: [63610973]; retrying with default start offset time
>> from configuration. configured start offset time: [-2]
>>
>> So why does the offset get reset?
>>
>>
>>> On 18 May 2015, at 20:37, Jeffery Maass <[email protected]
>>> <mailto:[email protected]>> wrote:
>>>
>>> The answer will be in how you configured the kafka spout. If after reading
>>> the below, you still need help, please grab the values for all of the
>>> settings mentioned in the doc and send them on.
>>>
>>> See this document about the Kafka Spout:
>>> http://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.2.0/Storm_UG_v22/Storm_UG_v22.pdf
>>>
>>> <http://docs.hortonworks.com/HDPDocuments/HDP2/HDP-2.2.0/Storm_UG_v22/Storm_UG_v22.pdf>
>>>
>>> See this archive message:
>>> http://mail-archives.apache.org/mod_mbox/storm-user/201503.mbox/%3ccahzwdygra33uiv+po01mvthvrjsxvqlvx6babymzye8zu_d...@mail.gmail.com%3E
>>>
>>> <http://mail-archives.apache.org/mod_mbox/storm-user/201503.mbox/%3ccahzwdygra33uiv+po01mvthvrjsxvqlvx6babymzye8zu_d...@mail.gmail.com%3E>
>>>
>>> "
>>> Not exactly.. forceFromStart=true will tell the spout to start reading from
>>> whatever is set in startOffsetTime (available options are the earliest
>>> offset or the latest offset). If forceFromStart=false then startOffsetTime
>>> is not used at all and the offset is just retrieved from zookeeper, if it's
>>> available.
>>>
>>> The "Start" in "forceFromStart" has nothing to do with consuming from the
>>> beginning of the topic. I interpret it as referring to whether you are
>>> going to force starting consumption from a different offset.
>>> "
>>>
>>> Thank you for your time!
>>>
>>> +++++++++++++++++++++
>>> Jeff Maass <mailto:[email protected]>
>>> linkedin.com/in/jeffmaass <http://linkedin.com/in/jeffmaass>
>>> stackoverflow.com/users/373418/maassql
>>> <http://stackoverflow.com/users/373418/maassql>
>>> +++++++++++++++++++++
>>>
>>>
>>> On Mon, May 18, 2015 at 1:53 PM, Benjamin Cuthbert <[email protected]
>>> <mailto:[email protected]>> wrote:
>>> All,
>>>
>>> We are getting loads of these errors
>>>
>>> 2015-05-18T19:52:44.038+0100 s.k.KafkaUtils [WARN] Got fetch request with
>>> offset out of range: [62379213]; retrying with default start offset time
>>> from configuration. configured start offset time: [-2]
>>> 2015-05-18T19:52:44.066+0100 s.k.PartitionManager [WARN] Using new offset:
>>> 63495047
>>>
>>> And it is stopping messages being pulled from Kafka into the spout and onto
>>> the bolts. Any ideas on how to fix this?
>>>
>>
>
