[jira] [Commented] (KAFKA-10197) Elect preferred leader when completing a partition reassignment

["Jun Rao (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-10197?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17144190#comment-17144190 ] 

Jun Rao commented on KAFKA-10197:
---------------------------------

[~bob-barrett]: Thanks for filing the issue. In reassignPartitionLeaderElection(), we do iterate the preferred replica first when selecting the leader.

{code:java}
  def reassignPartitionLeaderElection(reassignment: Seq[Int], isr: Seq[Int], liveReplicas: Set[Int]): Option[Int] = {
    reassignment.find(id => liveReplicas.contains(id) && isr.contains(id))
  }
{code}

The issue is that in KafkaController.moveReassignedPartitionLeaderIfRequired(), we won't trigger a leader election if the leader is in the reassigned replicas, but not the preferred one. 


{code:java}
  private def moveReassignedPartitionLeaderIfRequired(topicPartition: TopicPartition,
                                                      newAssignment: ReplicaAssignment): Unit = {
    val reassignedReplicas = newAssignment.replicas
    val currentLeader = controllerContext.partitionLeadershipInfo(topicPartition).get.leaderAndIsr.leader

    if (!reassignedReplicas.contains(currentLeader)) {
      info(s"Leader $currentLeader for partition $topicPartition being reassigned, " +
        s"is not in the new list of replicas ${reassignedReplicas.mkString(",")}. Re-electing leader")
      // move the leader to one of the alive and caught up new replicas
      partitionStateMachine.handleStateChanges(Seq(topicPartition), OnlinePartition, Some(ReassignPartitionLeaderElectionStrategy))
    } else if (controllerContext.isReplicaOnline(currentLeader, topicPartition)) {
      info(s"Leader $currentLeader for partition $topicPartition being reassigned, " +
        s"is already in the new list of replicas ${reassignedReplicas.mkString(",")} and is alive")
      // shrink replication factor and update the leader epoch in zookeeper to use on the next LeaderAndIsrRequest
      updateLeaderEpochAndSendRequest(topicPartition, newAssignment)
    } else {
      info(s"Leader $currentLeader for partition $topicPartition being reassigned, " +
        s"is already in the new list of replicas ${reassignedReplicas.mkString(",")} but is dead")
      partitionStateMachine.handleStateChanges(Seq(topicPartition), OnlinePartition, Some(ReassignPartitionLeaderElectionStrategy))
    }
  }
{code}





> Elect preferred leader when completing a partition reassignment
> ---------------------------------------------------------------
>
>                 Key: KAFKA-10197
>                 URL: https://issues.apache.org/jira/browse/KAFKA-10197
>             Project: Kafka
>          Issue Type: Improvement
>            Reporter: Bob Barrett
>            Priority: Major
>
> Currently, when completing a partition reassignment, we elect a leader from the new replica assignment without requiring that the leader be the new preferred leader. Instead, we just choose any in-sync replica:
> {code:java}
> private def leaderForReassign(partition: TopicPartition,
>                               leaderAndIsr: LeaderAndIsr,
>                               controllerContext: ControllerContext): ElectionResult = { 
>   val targetReplicas = controllerContext.partitionFullReplicaAssignment(partition).targetReplicaAssignment.replicas
>   val liveReplicas = targetReplicas.filter(replica => controllerContext.isReplicaOnline(replica, partition))
>   val isr = leaderAndIsr.isr
>   val leaderOpt = PartitionLeaderElectionAlgorithms.reassignPartitionLeaderElection(targetReplicas, isr, liveReplicas.toSet)
>   val newLeaderAndIsrOpt = leaderOpt.map(leader => leaderAndIsr.newLeader(leader, isUnclean = false))
>   ElectionResult(partition, newLeaderAndIsrOpt, targetReplicas)
> }
> {code}
> If auto preferred leader election is enabled, the preferred leader will eventually be elected. However, it would make sense to choose the preferred leader after the reassignment without waiting for the automatic trigger.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)