title: ReplicaFetcher crashing - First offset XXXXXXX is less than the next offset YYYYYYYY
body: <p>We have a 2.3 Kafka cluster of 3 boxes. When we did upgrade it to 2.3 few days ago, we notice those logs messages that make the replicaFetcher thread crash for one topic partition on two brokers:</p>
  
  <pre><code>[2019-08-09 15:02:43,520] ERROR [ReplicaFetcher replicaId=4, leaderId=3, fetcherId=0] Unexpected error occurred while processing data for partition __consumer_offsets-21 at offset 57542337 (kafka.server.R
  eplicaFetcherThread)
  kafka.common.UnexpectedAppendOffsetException: Unexpected offset in append to __consumer_offsets-21. First offset 57542333 is less than the next offset 57542337. First 10 offsets in append: List(57542333,
  57542334, 57542335, 57542336, 57542337, 57542338, 57542339, 57542340, 57542341, 57542342), last offset in append: 57570869. Log start offset = 56949140
          at kafka.log.Log.$anonfun$append$2(Log.scala:929)
          at kafka.log.Log.maybeHandleIOException(Log.scala:2065)
          at kafka.log.Log.append(Log.scala:850)
          at kafka.log.Log.appendAsFollower(Log.scala:830)
          at kafka.cluster.Partition.$anonfun$doAppendRecordsToFollowerOrFutureReplica$1(Partition.scala:726)
          at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:253)
          at kafka.utils.CoreUtils$.inReadLock(CoreUtils.scala:259)
          at kafka.cluster.Partition.doAppendRecordsToFollowerOrFutureReplica(Partition.scala:717)
          at kafka.cluster.Partition.appendRecordsToFollowerOrFutureReplica(Partition.scala:733)
          at kafka.server.ReplicaFetcherThread.processPartitionData(ReplicaFetcherThread.scala:161)
          at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$7(AbstractFetcherThread.scala:317)
          at scala.Option.foreach(Option.scala:274)
          at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6(AbstractFetcherThread.scala:306)
          at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$6$adapted(AbstractFetcherThread.scala:305)
          at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)
          at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)
          at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)
          at kafka.server.AbstractFetcherThread.$anonfun$processFetchRequest$5(AbstractFetcherThread.scala:305)
          at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
          at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:253)
          at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:305)
          at kafka.server.AbstractFetcherThread.$anonfun$maybeFetch$3(AbstractFetcherThread.scala:133)
          at kafka.server.AbstractFetcherThread.$anonfun$maybeFetch$3$adapted(AbstractFetcherThread.scala:132)
          at scala.Option.foreach(Option.scala:274)
          at kafka.server.AbstractFetcherThread.maybeFetch(AbstractFetcherThread.scala:132)
          at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:114)
          at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:89)
  [2019-08-09 15:02:43,524] WARN [ReplicaFetcher replicaId=4, leaderId=3, fetcherId=0] Partition __consumer_offsets-21 marked as failed (kafka.server.ReplicaFetcherThread)
  </code></pre>
  
  <p>The impact is that one broker is not able to be ISR for this topic partition ( actually the second broker has the same issue, so we only have one ISR which is the leader).</p>
  
  <p>I'm still confused about this message and I'm not able to understand it correctly, so I'm not able to find a proper way to get rid of this issue.
  I really want to understand what happen here, but not sure I understand the following code under the hood :</p>
  
  <p><a href="https://github.com/apache/kafka/blob/a48b5d900c6b5c9c52a97124a1b51aff3636c32c/core/src/main/scala/kafka/log/Log.scala#L1081-L1098" rel="nofollow noreferrer">https://github.com/apache/kafka/blob/a48b5d900c6b5c9c52a97124a1b51aff3636c32c/core/src/main/scala/kafka/log/Log.scala#L1081-L1098</a></p>
  
  <pre><code> if (appendInfo.firstOrLastOffsetOfFirstBatch &lt; nextOffsetMetadata.messageOffset) {
  </code></pre>
  
  <p>How can the replicaFetcher access the nextOffset information when it has to append records ..?
  Not sure to understand what this analyse does exactly( current records to append ?) :</p>
  
  <pre><code>val appendInfo = analyzeAndValidateRecords(records, isFromClient = isFromClient)
  </code></pre>
  
  <p>And this one :</p>
  
  <pre><code>nextOffsetMetadata
  </code></pre>
  
  <p>Is this the the next records batch.. ? How can it has access to whatever "next" records metadata ?</p>
  
  <p>If someone can clarify this a bit, would be great. 
  And in the meantime, a solution to get rid of the issue would be nice, but still, I would prefer understanding it clearly.</p>
  
  <p>EDITED:</p>
  
  <p>After some researches, some things got clearer.
  the nextOffset is just the latest offset+1 of the active segment ( those metadata comes from the loadSegments() call).</p>
  
  <p>So to sum up, here what happened:
  The replica get segment from the leader for which the beginning offsets are lower than the latest offset of the active segment. 
  So here my question, why the replica does not just truncate ?</p>
  
  <p>Yannick</p>