Repository: apache/kafka
File: core/src/main/scala/kafka/server/LocalLeaderEndPoint.scala (Scala)

Code before:
    new OffsetAndEpoch(logEndOffset, epoch.orElse(0))
  }

  override def fetchEarliestLocalOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int): OffsetAndEpoch = {
    val partition = replicaManager.getPartitionOrException(topicPartition)
    val localLogStartOffset = partition.localLogOrException.localLogStartOffset()
    val epoch = partition.localLogOrException.leaderEpochCache.epochForOffset(localLogStartOffset)
    new OffsetAndEpoch(localLogStartOffset, epoch.orElse(0))
  }

  override def fetchEarliestPendingUploadOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int): OffsetAndEpoch = {
    val partition = replicaManager.getPartitionOrException(topicPartition)
    val log = partition.localLogOrException

    if (!log.remoteLogEnabled())
      return new OffsetAndEpoch(-1L, -1)

    val highestRemoteOffset = log.highestOffsetInRemoteStorage()

    if (highestRemoteOffset == -1L) {
      val localLogStartOffset = fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch)
      val logStartOffset = fetchEarliestOffset(topicPartition, currentLeaderEpoch)

      if (localLogStartOffset.offset() == logStartOffset.offset()) {
        // No segments have been uploaded yet
        return logStartOffset;
      } else {
        // Leader currently does not know about the already uploaded segments
        return new OffsetAndEpoch(-1L, -1);
      }
    }

    val logStartOffset = fetchEarliestOffset(topicPartition, currentLeaderEpoch)
    val earliestPendingUploadOffset = Math.max(highestRemoteOffset + 1, logStartOffset.offset())
    val epoch = log.leaderEpochCache.epochForOffset(earliestPendingUploadOffset)

    new OffsetAndEpoch(earliestPendingUploadOffset, epoch.orElse(0))
  }

  override def fetchEpochEndOffsets(partitions: util.Map[TopicPartition, OffsetForLeaderEpochRequestData.OffsetForLeaderPartition]): util.Map[TopicPartition, EpochEndOffset] = {
    partitions.asScala.map { case (tp, epochData) =>
      try {
        val endOffset = if (epochData.leaderEpoch == UNDEFINED_EPOCH) {
          new EpochEndOffset()
            .setPartition(tp.partition)
            .setErrorCode(Errors.NONE.code)
        } else {
          val partition = replicaManager.getPartitionOrException(tp)
          partition.lastOffsetForLeaderEpoch(
            currentLeaderEpoch = RequestUtils.getLeaderEpoch(epochData.currentLeaderEpoch),
            leaderEpoch = epochData.leaderEpoch,

Review comment (refactor):
can we refactor the code to avoid using multiple return statements? 

https://tpolecat.github.io/2014/05/09/return.html



Code after:
    new OffsetAndEpoch(logEndOffset, epoch.orElse(0))
  }

  override def fetchEarliestLocalOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int): OffsetAndEpoch = {
    val partition = replicaManager.getPartitionOrException(topicPartition)
    val localLogStartOffset = partition.localLogOrException.localLogStartOffset()
    val epoch = partition.localLogOrException.leaderEpochCache.epochForOffset(localLogStartOffset)
    new OffsetAndEpoch(localLogStartOffset, epoch.orElse(0))
  }

  override def fetchEarliestPendingUploadOffset(topicPartition: TopicPartition, currentLeaderEpoch: Int): OffsetAndEpoch = {
    val partition = replicaManager.getPartitionOrException(topicPartition)
    val log = partition.localLogOrException

    if (!log.remoteLogEnabled()) {
      new OffsetAndEpoch(-1L, -1)
    } else {
      val highestRemoteOffset = log.highestOffsetInRemoteStorage()
      val logStartOffset = fetchEarliestOffset(topicPartition, currentLeaderEpoch)

      highestRemoteOffset match {
        case -1L =>
          val localLogStartOffset = fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch)
          if (localLogStartOffset.offset() == logStartOffset.offset()) {
            // No segments have been uploaded yet
            logStartOffset
          } else {
            // Leader currently does not know about the already uploaded segments
            new OffsetAndEpoch(-1L, -1)
          }
        case _ =>
          val earliestPendingUploadOffset = Math.max(highestRemoteOffset + 1, logStartOffset.offset())
          val epoch = log.leaderEpochCache.epochForOffset(earliestPendingUploadOffset)
          new OffsetAndEpoch(earliestPendingUploadOffset, epoch.orElse(0))
      }
    }
  }

  override def fetchEpochEndOffsets(partitions: util.Map[TopicPartition, OffsetForLeaderEpochRequestData.OffsetForLeaderPartition]): util.Map[TopicPartition, EpochEndOffset] = {
    partitions.asScala.map { case (tp, epochData) =>
      try {
        val endOffset = if (epochData.leaderEpoch == UNDEFINED_EPOCH) {
          new EpochEndOffset()
            .setPartition(tp.partition)
            .setErrorCode(Errors.NONE.code)
        } else {
          val partition = replicaManager.getPartitionOrException(tp)
          partition.lastOffsetForLeaderEpoch(
            currentLeaderEpoch = RequestUtils.getLeaderEpoch(epochData.currentLeaderEpoch),
            leaderEpoch = epochData.leaderEpoch,
            fetchOnlyFromLeader = false)