title: Kafka Consumer reads up to when it started, then hangs forever
body: <p>I've set up a Kafka solution with 1 producer and 1 consumer and validated that all the connections are correct (I can produce messages and consume them). ZK Server &amp; Kakfa Server are started and stable.</p>
  
  <p>As described, my issue is that the Consumer will read messages just fine and from where it left off, but will only read messages that were created until before it started reading. After that, new messages will not be read until I kill the Consumer and restart him again.</p>
  
  <p>Relevant Consumer Scala code</p>
  
  <pre><code>  val consumer = Consumer.create(new ConsumerConfig(readConsumerPropertiesFromConfig))
    val filterSpec = new Whitelist("some-valid-topic")
  
    val stream: KafkaStream[String, String] =
      consumer.createMessageStreamsByFilter(filterSpec, 1, new StringDecoder, new StringDecoder).head
  
    log.info(s"Consumer started. Listening to topics [$filterSpec].")
  
    def read() = stream map digest
  </code></pre>
  
  <p>Where digest takes a MessageAndMetadata and has fun with it</p>
  
  <pre><code>def digest(messageAndMeta: MessageAndMetadata[String, String]) = {
      log.info(s"processing the message [$messageAndMeta]")
  </code></pre>
  
  <p>Properties are</p>
  
  <pre><code>properties.put("group.id", "default_consumer_group")
  properties.put("zookeeper.connect", "localhost:2181")
  properties.put("auto.offset.reset", "smallest")
  properties.put("consumer.timeout.ms", 2000)
  </code></pre>
  
  <p>A timeline of what I can reproduce with this</p>
  
  <ul>
  <li>Produce 5 messages</li>
  <li>Start Consumer</li>
  <li>Consumer reads the 5 messages</li>
  <li>Produce 15 more messages</li>
  <li>Consumer ignores new messages and hangs forever</li>
  <li>Kill and restart Consumer</li>
  <li>Consumer reads the 15 messages, and again hangs forever</li>
  </ul>
  
  <p>Any thoughts? Thanks.</p>