title: Storm KafkaSpout stopped to consume messages from Kafka Topic
body: <p>My problem is that Storm KafkaSpout stopped to consume messages from Kafka topic after a period of time. When debug is enabled in storm, I get the log file like this:</p>
  <blockquote>
  <p>2016-07-05 03:58:26.097 o.a.s.d.task [INFO] Emitting: packet_spout __metrics [#object[org.apache.storm.metric.api.IMetricsConsumer$TaskInfo 0x2c35b34f &quot;org.apache.storm.metric.api.IMetricsConsumer$TaskInfo@2c35b34f&quot;] [#object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x798f1e35 &quot;[__ack-count = {default=0}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x230867ec &quot;[__sendqueue = {sojourn_time_ms=0.0, write_pos=5411461, read_pos=5411461, overflow=0, arrival_rate_secs=0.0, capacity=1024, population=0}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x7cdec8eb &quot;[__complete-latency = {default=0.0}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x658fc59 &quot;[__skipped-max-spout = 0]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x3c1f3a50 &quot;[__receive = {sojourn_time_ms=4790.5, write_pos=2468305, read_pos=2468304, overflow=0, arrival_rate_secs=0.20874647740319383, capacity=1024, population=1}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x262d7906 &quot;[__skipped-inactive = 0]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x73648c7e &quot;[kafkaPartition = {Partition{host=slave103:9092, topic=packet, partition=12}/fetchAPICallCount=0, Partition{host=slave103:9092, topic=packet, partition=12}/fetchAPILatencyMax=null, Partition{host=slave103:9092, topic=packet, partition=12}/lostMessageCount=0, Partition{host=slave103:9092, topic=packet, partition=12}/fetchAPILatencyMean=null, Partition{host=slave103:9092, topic=packet, partition=12}/fetchAPIMessageCount=0}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x4e43df61 &quot;[kafkaOffset = {packet/totalLatestCompletedOffset=154305947, packet/partition_12/spoutLag=82472754, packet/totalEarliestTimeOffset=233919465, packet/partition_12/earliestTimeOffset=233919465, packet/partition_12/latestEmittedOffset=154307691, packet/partition_12/latestTimeOffset=236778701, packet/totalLatestEmittedOffset=154307691, packet/partition_12/latestCompletedOffset=154305947, packet/totalLatestTimeOffset=236778701, packet/totalSpoutLag=82472754}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x49fe816b &quot;[__transfer-count = {__ack_init=0, default=0, __metrics=0}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x63e2bdc0 &quot;[__fail-count = {}]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x3b17bb7b &quot;[__skipped-throttle = 1086120]&quot;] #object[org.apache.storm.metric.api.IMetricsConsumer$DataPoint 0x1315a68c &quot;[__emit-count = {__ack_init=0, default=0, __metrics=0}]&quot;]]]</p>
  <p>2016-07-05 03:58:55.042 o.a.s.d.executor [INFO] <b>Processing received message FOR -2 TUPLE</b>: source: __system:-1, stream: __tick, id: {}, [30]</p>
  <p>2016-07-05 03:59:25.042 o.a.s.d.executor [INFO] <b>Processing received message FOR -2 TUPLE</b>: source: __system:-1, stream: __tick, id: {}, [30]</p>
  <p>2016-07-05 03:59:25.946 o.a.s.d.executor [INFO] <b>Processing received message FOR -2 TUPLE</b>: source: __system:-1, stream: __metrics_tick, id: {}, [60]</p>
  </blockquote>
  <p>My test topology is really simple, One KafkaSpout and another Counter Bolt. When the topology works fine, the value between <b>FOR</b> and <b>TUPLE</b> is a positive number; when the topology stops to consume the message, the value becomes negative. so I'm curious about what causes the problem of <b>Processing received message FOR -2 TUPLE</b>, and how to fix this problem?</p>
  <p>By the way, my experiment environment is:</p>
  <blockquote>
  <p>OS: Red Hat Enterprise Linux Server release 7.0 (Maipo)<br />
  Kafka: 0.10.0.0<br />
  Storm: 1.0.1</p>
  </blockquote>