title: Lost message from the Kafka Topic
body: <p>While trying timestamp in <a href="https://kafka.apache.org/11/javadoc/org/apache/kafka/clients/producer/ProducerRecord.html" rel="nofollow noreferrer">ProducerRecord</a>; I found something weird. After sending few messages from the producer, I ran kafka-console-consumer.sh and verified that those messages are in the topic. I stopped the producer and waited for a minute. When I reran kafka-console-consumer.sh then it did not show the messages that I generated previously. I also added producer.flush() and producer.close() but the outcome was still the same.</p>
  
  <p>Now, when I stopped using timestamp field then everything worked fine which makes me believe that there is something finicky about messages with timestamp.</p>
  
  <p>I am using Kafka_2.11-2.0.0 (released on July 30, 2018)</p>
  
  <p>Following is the sample code.</p>
  
  <pre><code>import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.producer.Producer;
  import org.apache.kafka.clients.producer.ProducerConfig;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.apache.kafka.common.header.internal.RecordHeaders;
  import org.apache.kafka.common.serialization.StringSerializer;
  import java.util.Properties;
  import static java.lang.Thread.sleep;
  public class KafkaProducerSample{
      public static void main(String[] args){
          String kafkaHost="sample:port";
          String notificationTopic="test";
  
          Properties props = new Properties();
          props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.kafkaHost);
          props.put(ProducerConfig.ACKS_CONFIG, 1);
          props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
  
          Producer&lt;String, String&gt; producer = new KafkaProducer(props, new StringSerialize(), new StringSerializer);
  
          RecordHeaders recordHeaders = new RecordHeader();
          ProducerRecord&lt;String, String&gt; record = new ProducerRecord(notificationTopic, null, 1574443515L, sampleKey, SampleValue);
          producer.send(record);
          sleep(1000);
      }
  }
  </code></pre>
  
  <p>I run console consumer as following</p>
  
  <pre><code>$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap.server KAFKA_HOST:PORT --topic test --from-beginning
  
  #output after running producer
  test
  
  
  #output 5mins after shutting down producer
  </code></pre>