[jira] [Created] (STORM-647) KafkaUtils repeat fetch messages which offset is out of range

["Xin Wang (JIRA)" <ji...@apache.org>]

Xin Wang created STORM-647:
------------------------------

             Summary: KafkaUtils repeat fetch messages which offset is out of range
                 Key: STORM-647
                 URL: https://issues.apache.org/jira/browse/STORM-647
             Project: Apache Storm
          Issue Type: Bug
          Components: storm-kafka
    Affects Versions: 0.9.2-incubating, 0.9.3
            Reporter: Xin Wang
            Assignee: Xin Wang


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.

[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;
}
...



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)