title: Preventing message loss with Kafka High Level Consumer 0.8.x
body: <p>A typical kafka consumer looks like the following:</p>
  
  <p>kafka-broker ---> kafka-consumer ----> downstream-consumer like Elastic-Search</p>
  
  <p>And according to the documentation for <a href="https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example" rel="nofollow">Kafka High Level Consumer</a>:</p>
  
  <blockquote>
    <p>The ‘auto.commit.interval.ms’ setting is how often updates to the
    consumed offsets are written to ZooKeeper</p>
  </blockquote>
  
  <p>It seems that there can be message loss if the following two things happen:</p>
  
  <ol>
  <li>Offsets are committed just after some messages are retrieved from kafka brokers.</li>
  <li>Downstream consumers (say Elastic-Search) fail to process the most recent batch of messages OR the consumer process itself is killed.</li>
  </ol>
  
  <p>It would perhaps be most ideal if the offsets are <strong>not</strong> committed automatically based on a time interval but they are committed by an API. This would make sure that the kafka-consumer can signal the committing of offsets only after it receives an acknowledgement from the downstream-consumer that they have successfully consumed the messages. There could be some replay of messages (if kafka-consumer dies before committing offsets) but there would at least be no message loss.</p>
  
  <p>Please let me know if such an API exists in the High Level Consumer.</p>
  
  <p>Note: I am aware of the Low Level Consumer API in 0.8.x version of Kafka but I do not want to manage everything myself when all I need is just one simple API in High Level Consumer.</p>
  
  <p>Ref:</p>
  
  <ol>
  <li><a href="https://github.com/apache/kafka/blob/d5b43b19bb06e9cdc606312c8bcf87ed267daf44/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L408-L435" rel="nofollow">AutoCommitTask.run()</a>, look for <em>commitOffsetsAsync</em></li>
  <li><a href="https://github.com/apache/kafka/blob/5db2c99e15a4467489b495144815ac8334c2c5d4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java#L285-L293" rel="nofollow">SubscriptionState.allConsumed()</a></li>
  </ol>