title: Can produce to Kafka but cannot consume
body: <p>I'm using the Kafka JDK client ver <code>0.10.2.1</code> . I am able to produce simple messages to Kafka for a &quot;heartbeat&quot; test, but I cannot consume a message from that same topic using the sdk. I am able to consume that message when I go into the Kafka CLI, so I have confirmed the message is there. Here's the function I'm using to consume from my Kafka server, with the props - I pass the message I produced to the topic only after I have indeed confirmed the <code>produce()</code> was succesful, I can post that function later if requested:</p>
  <pre><code>private def consumeFromKafka(topic: String, expectedMessage: String): Boolean = {
      val props: Properties = initProps(&quot;consumer&quot;)
      val consumer = new KafkaConsumer[String, String](props)
      consumer.subscribe(List(topic).asJava)
      var readExpectedRecord = false
      try {
        val records = {
          val firstPollRecs = consumer.poll(MAX_POLLTIME_MS)
          // increase timeout and try again if nothing comes back the first time in case system is busy
          if (firstPollRecs.count() == 0) firstPollRecs else {
            logger.info(&quot;KafkaHeartBeat: First poll had 0 records- trying again - doubling timeout to &quot;
              + (MAX_POLLTIME_MS * 2)/1000 + &quot; sec.&quot;)
            consumer.poll(MAX_POLLTIME_MS * 2)
          }
        }
        records.forEach(rec =&gt; {
          if (rec.value() == expectedMessage) readExpectedRecord = true
        })
      } catch {
        case e: Throwable =&gt; //log error
      } finally {
        consumer.close()
      }
      readExpectedRecord
    }
  
  private def initProps(propsType: String): Properties = {
      val prop = new Properties()
      prop.put(&quot;bootstrap.servers&quot;, kafkaServer + &quot;:&quot; + kafkaPort)
  
      propsType match {
        case &quot;producer&quot; =&gt; {
          prop.put(&quot;key.serializer&quot;, &quot;org.apache.kafka.common.serialization.StringSerializer&quot;)
          prop.put(&quot;value.serializer&quot;, &quot;org.apache.kafka.common.serialization.StringSerializer&quot;)
          prop.put(&quot;acks&quot;, &quot;1&quot;)
          prop.put(&quot;producer.type&quot;, &quot;sync&quot;)
          prop.put(&quot;retries&quot;, &quot;3&quot;)
          prop.put(&quot;linger.ms&quot;, &quot;5&quot;)
        }
        case &quot;consumer&quot; =&gt; {
          prop.put(&quot;group.id&quot;, groupId)
          prop.put(&quot;enable.auto.commit&quot;, &quot;false&quot;)
          prop.put(&quot;auto.commit.interval.ms&quot;, &quot;1000&quot;)
          prop.put(&quot;session.timeout.ms&quot;, &quot;30000&quot;)
          prop.put(&quot;key.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;)
          prop.put(&quot;value.deserializer&quot;, &quot;org.apache.kafka.common.serialization.StringDeserializer&quot;)
          prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;)
          // poll just once, should only be one record for the heartbeat
          prop.put(&quot;max.poll.records&quot;, &quot;1&quot;)
        }
      }
      prop
    }
  </code></pre>
  <p>Now when I run the code, here's what it outputs in the console:</p>
  <blockquote>
  <p>13:04:21  - Discovered coordinator serverName:9092 (id: 2147483647</p>
  <p>rack: null) for group 0b8947e1-eb68-4af3-ac7b-be3f7c02e76e. 13:04:23</p>
  <p>INFO  o.a.k.c.c.i.ConsumerCoordinator - Revoking previously assigned</p>
  <p>partitions [] for group 0b8947e1-eb68-4af3-ac7b-be3f7c02e76e 13:04:24</p>
  <p>INFO  o.a.k.c.c.i.AbstractCoordinator - (Re-)joining group</p>
  <p>0b8947e1-eb68-4af3-ac7b-be3f7c02e76e 13:04:25 INFO</p>
  <p>o.a.k.c.c.i.AbstractCoordinator - Successfully joined group</p>
  <p>0b8947e1-eb68-4af3-ac7b-be3f7c02e76e with generation 1 13:04:26  INFO</p>
  <p>o.a.k.c.c.i.ConsumerCoordinator - Setting newly assigned partitions</p>
  <p>[HeartBeat_Topic.Service_5.2018-08-03.13_04_10.377-0] for group</p>
  <p>0b8947e1-eb68-4af3-ac7b-be3f7c02e76e 13:04:27  INFO</p>
  <p>c.p.p.l.util.KafkaHeartBeatUtil - KafkaHeartBeat: First poll had 0</p>
  <p>records- trying again - doubling timeout to 60 sec.</p>
  </blockquote>
  <p>And then nothing else, no errors thrown -so no records are polled. Does anyone have any idea what's preventing the 'consume' from happening? The subscriber seems to be successful, as I'm able to successfully call the listTopics and list partions no problem.</p>