title: Flink + Kafka: Why am I losing messages?
body: <p>I have written a very simple Flink streaming job which takes data from Kafka using a <code>FlinkKafkaConsumer082</code>. </p>
  
  <pre><code>protected DataStream&lt;String&gt; getKafkaStream(StreamExecutionEnvironment env, String topic) {
      Properties result = new Properties();
      result.put("bootstrap.servers", getBrokerUrl());
      result.put("zookeeper.connect", getZookeeperUrl());
      result.put("group.id", getGroup());
  
          return env.addSource(
                  new FlinkKafkaConsumer082&lt;&gt;(
                          topic,
                          new SimpleStringSchema(), result);
  }
  </code></pre>
  
  <p>This works very well and whenever I put something into the topic on Kafka, it is received by my Flink job and processed. Now I tried to see what happens if my Flink Job isn't online for some reason. So I shut down the flink job and kept sending messages to Kafka. Then I started my Flink job again and was expecting that it would process the messages that were sent meanwhile.</p>
  
  <p>However, I got this message:</p>
  
  <pre><code>No prior offsets found for some partitions in topic collector.Customer. Fetched the following start offsets [FetchPartition {partition=0, offset=25}]
  </code></pre>
  
  <p>So it basically ignored all messages that came since the last shutdown of the Flink job and just started to read at the end of the queue. From the documentation of <code>FlinkKafkaConsumer082</code> I gathered, that it automatically takes care of synchronizing the processed offsets with the Kafka broker. However that doesn't seem to be the case.</p>
  
  <p>I am using a single-node Kafka installation (the one that comes with the Kafka distribution) with a single-node Zookeper installation (also the one that is bundled with the Kafka distribution).</p>
  
  <p>I suspect it is some kind of misconfiguration or something the like but I really don't know where to start looking. Has anyone else had this issue and maybe solved it? </p>