wurstmeister/storm-kafka-0.8-plus — pull request #3 by EdisonXu (2013-12-17 01:30:40+00:00)
Title: Trying to fix the 1st time launch limitation

In the PartitionManager, if it's the first time launch,there's no any JSON data under the commit path of zk. Thus, it will read the latest Offset from the kafka.
Let's say the kafka is ready long ago, and there's some producer who just starts a little bit earilier than storm topology, which means some data will be pushed into kafka.
If so, those kinds of data will be lost since PartitionManager just picks the latest offset.

I suggest add a following
if(json==null) committedTo = KafkaUtils.getOffset(consumer, spoutConfig.topic, id.partition, kafka.api.OffsetRequest.EarliestTime());
right after the line getting JSON data from commit path.

By doing so, it will pick the earliest data, so no data will be lost. Indeed, if someone deletes the commit path in the zk, it will read duplicate data. But that won't happen in production, right?