[STORM-643] KafkaUtils repeatedly fetches messages whose offset is out of range

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.9.2-incubating, 0.9.3, 0.10.0, 0.9.4, 0.9.5
• Fix Version/s: 0.9.6
• Component/s: storm-kafka
• Assignee: Xin Wang
• Reporter: Xin Wang
• Due: 01/Feb/15
• Created: 01/Feb/15 09:20
• Updated: 20/Jun/16 03:38
• Resolved: 20/Jul/15 15:37

Description

KafkaUtils repeat fetch messages which offset is out of range.
This happened when failed list(SortedSet<Long> failed) is not empty and some offset in it is OutOfRange.

[worker-log]

2015-02-01 10:24:27.231+0800 s.k.KafkaUtils [WARN] Got fetch request with offset out of range: [20919071816]; retrying with default start offset time from configuration. configured start offset time: [-2]
2015-02-01 10:24:27.232+0800 s.k.PartitionManager [WARN] Using new offset: 20996130717
2015-02-01 10:24:27.333+0800 s.k.KafkaUtils [WARN] Got fetch request with offset out of range: [20919071816]; retrying with default start offset time from configuration. configured start offset time: [-2]
2015-02-01 10:24:27.334+0800 s.k.PartitionManager [WARN] Using new offset: 20996130717
...

[FIX]

storm.kafka.PartitionManager.fill():
...
try {
	msgs = KafkaUtils.fetchMessages(_spoutConfig, _consumer, _partition, offset);
} catch (UpdateOffsetException e) {
	 _emittedToOffset = KafkaUtils.getOffset(_consumer, _spoutConfig.topic, _partition.partition, _spoutConfig);
	LOG.warn("Using new offset: {}", _emittedToOffset);
	// fetch failed, so don't update the metrics

	//fix bug: remove this offset from failed list when it is OutOfRange
	if (had_failed) {
		failed.remove(offset);
	}

            return;
}
...

also: Log "retrying with default start offset time from configuration. configured start offset time: [-2]" is incorrect.

Issue Links

• duplicates: STORM-394 Messages has expired, OFFSET_OUT_OF_RANGE, new offset startOffsetTime, no new messages, again and again
• duplicates: STORM-644 KafkaUtils repeat fetch messages which offset is out of range
• duplicates: STORM-645 KafkaUtils repeat fetch messages which offset is out of range
• duplicates: STORM-646 KafkaUtils repeat fetch messages which offset is out of range
• duplicates: STORM-647 KafkaUtils repeat fetch messages which offset is out of range
• duplicates: STORM-732 PartitionManager should remove those messages not exists in kafka from waiting queue
• is related to: STORM-586 Trident kafka spout fails instead of updating offset when kafka offset is out of range.
• is related to: STORM-511 Storm-Kafka spout keeps sending fetch requests with invalid offset