title: Kafka: Intermittent slowness when consuming first message from topic
body: <p>I am using Kafka 0.9.0.1.</p>
  
  <p>The first time I start up my application it takes 20-30 seconds to retrieve the "latest" message from the topic</p>
  
  <p>I've used different Kafka brokers (with different configs) yet I still see this behaviour. There is usually no slowness for subsequent messages.</p>
  
  <p>Is this expected behaviour? you can clearly see this below by running this sample application and changing the broker/topic name to your own settings</p>
  
  <pre><code>public class KafkaProducerConsumerTest {
  
      public static final String KAFKA_BROKERS = "...";
      public static final String TOPIC = "...";
  
      public static void main(String[] args) throws ExecutionException, InterruptedException {
          new KafkaProducerConsumerTest().run();
      }
  
      public void run() throws ExecutionException, InterruptedException {
          Properties consumerProperties = new Properties();
          consumerProperties.setProperty("bootstrap.servers", KAFKA_BROKERS);
          consumerProperties.setProperty("group.id", "Test");
          consumerProperties.setProperty("auto.offset.reset", "latest");
          consumerProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          consumerProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          MyKafkaConsumer kafkaConsumer = new MyKafkaConsumer(consumerProperties, TOPIC);
          Executors.newFixedThreadPool(1).submit(() -&gt; kafkaConsumer.consume());
  
          Properties producerProperties = new Properties();
          producerProperties.setProperty("bootstrap.servers", KAFKA_BROKERS);
          producerProperties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
          producerProperties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  
          MyKafkaProducer kafkaProducer = new MyKafkaProducer(producerProperties, TOPIC);
          kafkaProducer.publish("Test Message");
      }
  }
  
  
  class MyKafkaConsumer {
      private final Logger logger = LoggerFactory.getLogger(MyKafkaConsumer.class);
      private KafkaConsumer&lt;String, Object&gt; kafkaConsumer;
  
      public MyKafkaConsumer(Properties properties, String topic) {
          kafkaConsumer = new KafkaConsumer&lt;String, Object&gt;(properties);
          kafkaConsumer.subscribe(Lists.newArrayList(topic));
      }
  
      public void consume() {
          while (true) {
              logger.info("Started listening...");
              ConsumerRecords&lt;String, Object&gt; consumerRecords = kafkaConsumer.poll(Long.MAX_VALUE);
              logger.info("Received records {}", consumerRecords.iterator().next().value());
          }
      }
  }
  
  class MyKafkaProducer {
      private KafkaProducer&lt;String, Object&gt; kafkaProducer;
      private String topic;
  
      public MyKafkaProducer(Properties properties, String topic) {
          this.kafkaProducer = new KafkaProducer&lt;String, Object&gt;(properties);
          this.topic = topic;
      }
  
      public void publish(Object object) throws ExecutionException, InterruptedException {
          ProducerRecord&lt;String, Object&gt; producerRecord = new ProducerRecord&lt;&gt;(topic, "key", object);
          Future&lt;RecordMetadata&gt; response = kafkaProducer.send(producerRecord);
          response.get();
      }
  
  }
  </code></pre>