title: Consistency of kafka consumer endOffsets
body: <p>I have kafka topic with replication factor 3 and <code>min.insync.replicas = 2</code>, a producer that sends X messages to this topic with <code>acks=all</code>.
  After a while (within 1 min) after all messages sent to the topic new consumer is created for this topic using java kafka client. Using <code>consumer.endOffsets()</code> method end offsets of all kafka partitions for this topic are fetched. Another invocation of the same method <code>consumer.endOffsets</code> sometimes return different end offsets for some partitions.</p>
  
  <p>In this setup there are no new messages sent to kafka topic after the creation of the consumer.</p>
  
  <p>According to java docs of <code>endOffsets</code>:</p>
  
  <pre><code>/**
       * Get the last offset for the given partitions.  The last offset of a partition is the offset of the upcoming
       * message, i.e. the offset of the last available message + 1.  If messages have never been written
       * to the the partition, the offset returned will be 0.
       *
       * &lt;p&gt;
       * This method does not change the current consumer position of the partitions.
       * &lt;p&gt;
       * When {@code isolation.level=read_committed} the last offset will be the Last Stable Offset (LSO).
       * This is the offset of the first message with an open transaction. The LSO moves forward as transactions
       * are completed.
       *
       * @see #seekToEnd(Collection)
       *
       * @param partitions the partitions to get the end offsets.
       * @return The end offsets for the given partitions.
       * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details
       * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic(s). See the exception for more details
       * @throws org.apache.kafka.common.errors.TimeoutException if the offsets could not be fetched before
       *         expiration of the configured {@code request.timeout.ms}
       */
      @Override
      public Map&lt;TopicPartition, Long&gt; endOffsets(Collection&lt;TopicPartition&gt; partitions) 
  </code></pre>
  
  <p>endOffsets returns Last Stable Offset (LSO) which acknowledges by all replicas. </p>
  
  <p>How come that sometimes (not very often) end offsets changing between subsequent invocation of this method? 
  Is it expected behavior that endOffsets are eventually consistent? A Bug?</p>