大厂面试Kafka，一定会问到的幂等性

private def maybeValidateAppend(producerEpoch: Short, firstSeq: Int, offset: Long): Unit = {
    validationType match {
      case ValidationType.None =>
​
      case ValidationType.EpochOnly =>
        checkProducerEpoch(producerEpoch, offset)
​
      case ValidationType.Full =>
        checkProducerEpoch(producerEpoch, offset)
        checkSequence(producerEpoch, firstSeq, offset)
    }
}
​
private def checkSequence(producerEpoch: Short, appendFirstSeq: Int, offset: Long): Unit = {
  if (producerEpoch != updatedEntry.producerEpoch) {
    if (appendFirstSeq != ) {
      if (updatedEntry.producerEpoch != RecordBatch.NO_PRODUCER_EPOCH) {
        throw new OutOfOrderSequenceException(s"Invalid sequence number for new epoch at offset $offset in " +
          s"partition $topicPartition: $producerEpoch (request epoch), $appendFirstSeq (seq. number)")
      } else {
        throw new UnknownProducerIdException(s"Found no record of producerId=$producerId on the broker at offset $offset" +
          s"in partition $topicPartition. It is possible that the last message with the producerId=$producerId has " +
          "been removed due to hitting the retention limit.")
      }
    }
  } else {
    val currentLastSeq = if (!updatedEntry.isEmpty)
      updatedEntry.lastSeq
    else if (producerEpoch == currentEntry.producerEpoch)
      currentEntry.lastSeq
    else
      RecordBatch.NO_SEQUENCE
​
    if (currentLastSeq == RecordBatch.NO_SEQUENCE && appendFirstSeq != ) {
      // We have a matching epoch, but we do not know the next sequence number. This case can happen if
      // only a transaction marker is left in the log for this producer. We treat this as an unknown
      // producer id error, so that the producer can check the log start offset for truncation and reset
      // the sequence number. Note that this check follows the fencing check, so the marker still fences
      // old producers even if it cannot determine our next expected sequence number.
      throw new UnknownProducerIdException(s"Local producer state matches expected epoch $producerEpoch " +
        s"for producerId=$producerId at offset $offset in partition $topicPartition, but the next expected " +
        "sequence number is not known.")
    } else if (!inSequence(currentLastSeq, appendFirstSeq)) {
      throw new OutOfOrderSequenceException(s"Out of order sequence number for producerId $producerId at " +
        s"offset $offset in partition $topicPartition: $appendFirstSeq (incoming seq. number), " +
        s"$currentLastSeq (current end sequence number)")
    }
  }
}
​
  private def inSequence(lastSeq: Int, nextSeq: Int): Boolean = {
    nextSeq == lastSeq + 1L || (nextSeq ==  && lastSeq == Int.MaxValue)
  }