title: storm-kafka Spout is retrieving the wrong offset
body: <p>Using kafka 0.8.1.1 with their respective zookeeper and the current code of kafka-storm in <a href="https://github.com/apache/incubator-storm(last" rel="nofollow">https://github.com/apache/incubator-storm(last</a> commit: 7ac24b893c) I had the following error:</p>
  
  <blockquote>
    <p>If I create two topics(topicA and topicB) and send data to
    topicA(changing the offset of topicA to offsetA) and open a kafka
    spout for topicA, everything is ok. But, if I open a spout for topicB,
    the spout is retrieving the offset of the topicA, leading to a
    Exception related to this wrong offset(offsetA instead of offsetB).</p>
  </blockquote>
  
  <p>Inspecting the code in: <a href="https://github.com/apache/incubator-storm/blob/master/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java#L103" rel="nofollow">https://github.com/apache/incubator-storm/blob/master/external/storm-kafka/src/jvm/storm/kafka/PartitionManager.java#L103</a>, is retrieving the correct offsetB, but instead of using this variable, is using an offset found in a json. This json, is found in <strong>/id/partition_0</strong>(a zookeeper url) where both topics are being stored.</p>
  
  <p>If I get the data from partition_0 from zookeeper, I receive the following information(beautified for readibility):</p>
  
  <pre><code>$ get /id/partition_0
  {
  "topology": {
      "id": "58e1c603-0ab0-4344-a614-204e4455e90a",
      "name": "Start test storm"
  },
  "offset": 1,
  "partition": 0,
  "broker": {
      "host": "XXX.XXX.XXX.XXX.compute.amazonaws.com",
      "port": 9092
  },
  "topic": "topicA"
  }
   ...
  </code></pre>
  
  <p>As you can see, is giving the offset of topicA. This is the behaviour that I don't understand: If I'm retrieving the offset from the same url(<strong>/id/partition_0</strong>) for both topics, is obvious that for some topic, a wrong offset will be retrieved. Is something misconfigured in my zookeeper-kafka configuration? Is other the cause of this error?</p>
  
  <p>I modified the code in that class(PartitionManager constructor) in order to ignore that json from <strong>/id/partition_0</strong> and using the offset from <strong>KafkaUtils.getOffset(_consumer, spoutConfig.topic, id.partition, spoutConfig);</strong> and my dataflow runs correctly.</p>
  
  <p>If helps, I will also add the zookeeper information for both topics.</p>
  
  <pre><code>$ get /brokers/topics/topicA
  {"version":1,"partitions":{"0":[0]}}
  ...
  
  
  $ get /brokers/topics/topicB
  {"version":1,"partitions":{"0":[0]}}
  ...
  </code></pre>