MINOR: Refactor replica log dir fetching for improved logging

[hachikuji]

In order to debug problems with log directory reassignments, it is helpful to know when the fetcher thread begins moving a particular partition. This patch refactors the fetch logic so that we stick to a selected partition as long as it is available and log a message when a different partition is selected.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[hachikuji]

cc @lindong28 or @junrao

Let me know if this seems reasonable.

[lindong28]

Hey @hachikuji, the motivation makes sense. I personally find it that the code looks a bit complicated as we split the original method into four separate methods. Other than that the patch LGTM. I will leave Jun or other people to decide the style.

[hachikuji]

@lindong28 Thanks for the comment. The most important thing is whether the change itself makes sense.

I agree that style leaves a lot of room for subjectivity. I personally find that smaller functions which are understandable in isolation lead to simpler and clearer composition of logic (and often better testing). For example, compare the original here:

  def buildFetch(partitionMap: Map[TopicPartition, PartitionFetchState]): ResultWithPartitions[Option[FetchRequest.Builder]] = {
    // Only include replica in the fetch request if it is not throttled.
    val maxPartitionOpt = partitionMap.filter { case (_, partitionFetchState) =>
      partitionFetchState.isReadyForFetch && !quota.isQuotaExceeded
    }.reduceLeftOption { (left, right) =>
      if ((left._1.topic > right._1.topic()) || (left._1.topic == right._1.topic() && left._1.partition() >= right._1.partition()))
        left
      else
        right
    }

    // Only move one replica at a time to increase its catch-up rate and thus reduce the time spent on moving any given replica
    // Replicas are ordered by their TopicPartition
    val requestMap = new util.LinkedHashMap[TopicPartition, FetchRequest.PartitionData]
    val partitionsWithError = mutable.Set[TopicPartition]()

    if (maxPartitionOpt.nonEmpty) {
      val (topicPartition, partitionFetchState) = maxPartitionOpt.get
      try {
        val logStartOffset = replicaMgr.futureLocalReplicaOrException(topicPartition).logStartOffset
        requestMap.put(topicPartition, new FetchRequest.PartitionData(partitionFetchState.fetchOffset, logStartOffset,
          fetchSize, Optional.of(partitionFetchState.currentLeaderEpoch)))
      } catch {
        case _: KafkaStorageException =>
          partitionsWithError += topicPartition
      }
    }

    val fetchRequestOpt = if (requestMap.isEmpty) {
      None
    } else {
      // Set maxWait and minBytes to 0 because the response should return immediately if
      // the future log has caught up with the current log of the partition
      Some(FetchRequest.Builder.forReplica(ApiKeys.FETCH.latestVersion, replicaId, 0, 0, requestMap)
        .setMaxBytes(maxBytes))
    }
    ResultWithPartitions(fetchRequestOpt, partitionsWithError)
}

with the update:

  def buildFetch(partitionMap: Map[TopicPartition, PartitionFetchState]): ResultWithPartitions[Option[FetchRequest.Builder]] = {
    // Only include replica in the fetch request if it is not throttled.
    if (quota.isQuotaExceeded) {
      ResultWithPartitions(None, Set.empty)
    } else {
      selectPartitionToFetch(partitionMap) match {
        case Some((tp, fetchState)) =>
          buildFetchForPartition(tp, fetchState)
        case None =>
          ResultWithPartitions(None, Set.empty)
      }
    }
  }

I didn't have too much trouble understanding the logic from the original version, but I think this version is much more approachable for contributors who haven't spent as much time working with Kafka as you and I.

[viktorsomogyi]

Reviewed it, LGTM.

[lindong28]

@hachikuji Yeah I am not strong on this preference. Just from my personal opinion, for most developers who do not have to change the implementation of buildFetch(...), they only need to read the Java doc of this method and thus there is no difference between these two implementations.

If the developer needs to understand/change the implementation of this method, then the developer needs to understand how the input is actually used to derive the output. In this case, even if buildFetch(...) itself can be much shorter by using 3 more private methods, the developer still needs to jump to the definition of each method to understand their implementation, which seems less convenient then having these in the same method. If the private method is short and not re-used more than once, then in many cases I find it simpler to inline the implementation in the original method with clear comments. This is the experience I have had when reviewing this patch, since I can not simply read the buildFetch(...) itself to be sure whether this is correct.

This can be just my personal preference regarding the style. The change itself looks good to me.

[junrao]

@hachikuji : Thanks for the PR. LGTM. Just a couple of minor comments below.

[junrao]

unavailable => unavailable or is removed ?

[junrao]

for largest partition => for first partition ?