分区副本重分配源码原理分析 (附配套教学视频)

  def generateAssignment(zkClient: KafkaZkClient, brokerListToReassign: Seq[Int], topicsToMoveJsonString: String, disableRackAware: Boolean): (Map[TopicPartition, Seq[Int]], Map[TopicPartition, Seq[Int]]) = {    //解析出游哪些Topic    val topicsToReassign = parseTopicsData(topicsToMoveJsonString)    //检查是否有重复的topic    val duplicateTopicsToReassign = CoreUtils.duplicates(topicsToReassign)    if (duplicateTopicsToReassign.nonEmpty)      throw new AdminCommandFailedException("List of topics to reassign contains duplicate entries: %s".format(duplicateTopicsToReassign.mkString(",")))      //获取topic当前的副本分配情况 /brokers/topics/{topicName}    val currentAssignment = zkClient.getReplicaAssignmentForTopics(topicsToReassign.toSet)
    val groupedByTopic = currentAssignment.groupBy { case (tp, _) => tp.topic }    //机架感知模式    val rackAwareMode = if (disableRackAware) RackAwareMode.Disabled else RackAwareMode.Enforced    val adminZkClient = new AdminZkClient(zkClient)    val brokerMetadatas = adminZkClient.getBrokerMetadatas(rackAwareMode, Some(brokerListToReassign))
    val partitionsToBeReassigned = mutable.Map[TopicPartition, Seq[Int]]()    groupedByTopic.foreach { case (topic, assignment) =>      val (_, replicas) = assignment.head      val assignedReplicas = AdminUtils.assignReplicasToBrokers(brokerMetadatas, assignment.size, replicas.size)      partitionsToBeReassigned ++= assignedReplicas.map { case (partition, replicas) =>        new TopicPartition(topic, partition) -> replicas      }    }    (partitionsToBeReassigned, currentAssignment)  }

  def executeAssignment(zkClient: KafkaZkClient, adminClientOpt: Option[Admin], reassignmentJsonString: String, throttle: Throttle, timeoutMs: Long = 10000L): Unit = {    //对json文件进行校验和解析    val (partitionAssignment, replicaAssignment) = parseAndValidate(zkClient, reassignmentJsonString)    val adminZkClient = new AdminZkClient(zkClient)    val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, adminClientOpt, partitionAssignment.toMap, replicaAssignment, adminZkClient)
     //检查是否已经存在副本重分配进程, 则尝试限流     if (zkClient.reassignPartitionsInProgress()) {            reassignPartitionsCommand.maybeLimit(throttle)    } else {      //打印当前的副本分配方式,方便回滚      printCurrentAssignment(zkClient, partitionAssignment.map(_._1.topic))      if (throttle.interBrokerLimit >= 0 || throttle.replicaAlterLogDirsLimit >= 0)        println(String.format("Warning: You must run Verify periodically, until the reassignment completes, to ensure the throttle is removed. You can also alter the throttle by rerunning the Execute command passing a new value."))        //开始进行重分配进程      if (reassignPartitionsCommand.reassignPartitions(throttle, timeoutMs)) {        println("Successfully started reassignment of partitions.")      } else        println("Failed to reassign partitions %s".format(partitionAssignment))    }  }

  def maybeLimit(throttle: Throttle): Unit = {    if (throttle.interBrokerLimit >= 0 || throttle.replicaAlterLogDirsLimit >= 0) {      //当前存在的broker      val existingBrokers = existingAssignment().values.flatten.toSeq      //期望的broker      val proposedBrokers = proposedPartitionAssignment.values.flatten.toSeq ++ proposedReplicaAssignment.keys.toSeq.map(_.brokerId())      //前面broker相加去重      val brokers = (existingBrokers ++ proposedBrokers).distinct
      //遍历与之相关的Brokers, 添加限流配置写入到zk节点/config/broker/{brokerId}中       for (id <- brokers) {        //获取broker的配置 /config/broker/{brokerId}        val configs = adminZkClient.fetchEntityConfig(ConfigType.Broker, id.toString)        if (throttle.interBrokerLimit >= 0) {          configs.put(DynamicConfig.Broker.LeaderReplicationThrottledRateProp, throttle.interBrokerLimit.toString)          configs.put(DynamicConfig.Broker.FollowerReplicationThrottledRateProp, throttle.interBrokerLimit.toString)        }        if (throttle.replicaAlterLogDirsLimit >= 0)          configs.put(DynamicConfig.Broker.ReplicaAlterLogDirsIoMaxBytesPerSecondProp, throttle.replicaAlterLogDirsLimit.toString)
        adminZkClient.changeBrokerConfig(Seq(id), configs)      }    }  }

{"version":1,"config":{"leader.replication.throttled.rate":"1","follower.replication.throttled.rate":"1"}}

  def reassignPartitions(throttle: Throttle = NoThrottle, timeoutMs: Long = 10000L): Boolean = {    //写入一些限流数据    maybeThrottle(throttle)    try {     //验证分区是否存在      val validPartitions = proposedPartitionAssignment.groupBy(_._1.topic())        .flatMap { case (topic, topicPartitionReplicas) =>          validatePartition(zkClient, topic, topicPartitionReplicas)        }      if (validPartitions.isEmpty) false      else {        if (proposedReplicaAssignment.nonEmpty && adminClientOpt.isEmpty)          throw new AdminCommandFailedException("bootstrap-server needs to be provided in order to reassign replica to the specified log directory")        val startTimeMs = System.currentTimeMillis()
        // Send AlterReplicaLogDirsRequest to allow broker to create replica in the right log dir later if the replica has not been created yet.        if (proposedReplicaAssignment.nonEmpty)          alterReplicaLogDirsIgnoreReplicaNotAvailable(proposedReplicaAssignment, adminClientOpt.get, timeoutMs)
        // Create reassignment znode so that controller will send LeaderAndIsrRequest to create replica in the broker        zkClient.createPartitionReassignment(validPartitions.map({case (key, value) => (new TopicPartition(key.topic, key.partition), value)}).toMap)
        // Send AlterReplicaLogDirsRequest again to make sure broker will start to move replica to the specified log directory.        // It may take some time for controller to create replica in the broker. Retry if the replica has not been created.        var remainingTimeMs = startTimeMs + timeoutMs - System.currentTimeMillis()        val replicasAssignedToFutureDir = mutable.Set.empty[TopicPartitionReplica]        while (remainingTimeMs > 0 && replicasAssignedToFutureDir.size < proposedReplicaAssignment.size) {          replicasAssignedToFutureDir ++= alterReplicaLogDirsIgnoreReplicaNotAvailable(            proposedReplicaAssignment.filter { case (replica, _) => !replicasAssignedToFutureDir.contains(replica) },            adminClientOpt.get, remainingTimeMs)          Thread.sleep(100)          remainingTimeMs = startTimeMs + timeoutMs - System.currentTimeMillis()        }        replicasAssignedToFutureDir.size == proposedReplicaAssignment.size      }    } catch {      case _: NodeExistsException =>        val partitionsBeingReassigned = zkClient.getPartitionReassignment()        throw new AdminCommandFailedException("Partition reassignment currently in " +          "progress for %s. Aborting operation".format(partitionsBeingReassigned))    }  }

{"version":1,"partitions":[{"topic":"Topic1","partition":2,"replicas":[1],"log_dirs":["/Users/shirenchuang/work/IdeaPj/didi_source/kafka/k0"]}]}

private def processZkPartitionReassignment(): Set[TopicPartition] = {    // We need to register the watcher if the path doesn't exist in order to detect future    // reassignments and we get the `path exists` check for free    if (isActive && zkClient.registerZNodeChangeHandlerAndCheckExistence(partitionReassignmentHandler)) {      val reassignmentResults = mutable.Map.empty[TopicPartition, ApiError]      val partitionsToReassign = mutable.Map.empty[TopicPartition, ReplicaAssignment]
      zkClient.getPartitionReassignment().foreach { case (tp, targetReplicas) =>        maybeBuildReassignment(tp, Some(targetReplicas)) match {          case Some(context) => partitionsToReassign.put(tp, context)          case None => reassignmentResults.put(tp, new ApiError(Errors.NO_REASSIGNMENT_IN_PROGRESS))        }      }
      reassignmentResults ++= maybeTriggerPartitionReassignment(partitionsToReassign)      val (partitionsReassigned, partitionsFailed) = reassignmentResults.partition(_._2.error == Errors.NONE)      if (partitionsFailed.nonEmpty) {        warn(s"Failed reassignment through zk with the following errors: $partitionsFailed")        maybeRemoveFromZkReassignment((tp, _) => partitionsFailed.contains(tp))      }      partitionsReassigned.keySet    } else {      Set.empty    }  }

private def onPartitionReassignment(topicPartition: TopicPartition, reassignment: ReplicaAssignment): Unit = {    // 暂停一些正在删除的Topic操作   topicDeletionManager.markTopicIneligibleForDeletion(Set(topicPartition.topic), reason = "topic reassignment in progress")    //更新当前的分配    updateCurrentReassignment(topicPartition, reassignment)
    val addingReplicas = reassignment.addingReplicas    val removingReplicas = reassignment.removingReplicas
    if (!isReassignmentComplete(topicPartition, reassignment)) {      // A1. Send LeaderAndIsr request to every replica in ORS + TRS (with the new RS, AR and RR).      updateLeaderEpochAndSendRequest(topicPartition, reassignment)      // A2. replicas in AR -> NewReplica      startNewReplicasForReassignedPartition(topicPartition, addingReplicas)    } else {      // B1. replicas in AR -> OnlineReplica      replicaStateMachine.handleStateChanges(addingReplicas.map(PartitionAndReplica(topicPartition, _)), OnlineReplica)      // B2. Set RS = TRS, AR = [], RR = [] in memory.      val completedReassignment = ReplicaAssignment(reassignment.targetReplicas)      controllerContext.updatePartitionFullReplicaAssignment(topicPartition, completedReassignment)      // B3. Send LeaderAndIsr request with a potential new leader (if current leader not in TRS) and      //   a new RS (using TRS) and same isr to every broker in ORS + TRS or TRS      moveReassignedPartitionLeaderIfRequired(topicPartition, completedReassignment)      // B4. replicas in RR -> Offline (force those replicas out of isr)      // B5. replicas in RR -> NonExistentReplica (force those replicas to be deleted)      stopRemovedReplicasOfReassignedPartition(topicPartition, removingReplicas)      // B6. Update ZK with RS = TRS, AR = [], RR = [].      updateReplicaAssignmentForPartition(topicPartition, completedReassignment)      // B7. Remove the ISR reassign listener and maybe update the /admin/reassign_partitions path in ZK to remove this partition from it.      removePartitionFromReassigningPartitions(topicPartition, completedReassignment)      // B8. After electing a leader in B3, the replicas and isr information changes, so resend the update metadata request to every broker      sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(topicPartition))      // signal delete topic thread if reassignment for some partitions belonging to topics being deleted just completed      topicDeletionManager.resumeDeletionForTopics(Set(topicPartition.topic))    }  }

  private def processPartitionModifications(topic: String): Unit = {    def restorePartitionReplicaAssignment(      topic: String,      newPartitionReplicaAssignment: Map[TopicPartition, ReplicaAssignment]    ): Unit = {      info("Restoring the partition replica assignment for topic %s".format(topic))
      //从zk节点中获取所有分区      val existingPartitions = zkClient.getChildren(TopicPartitionsZNode.path(topic))      //找到已经存在的分区      val existingPartitionReplicaAssignment = newPartitionReplicaAssignment        .filter(p => existingPartitions.contains(p._1.partition.toString))        .map { case (tp, _) =>          tp -> controllerContext.partitionFullReplicaAssignment(tp)      }.toMap
      zkClient.setTopicAssignment(topic,        existingPartitionReplicaAssignment,        controllerContext.epochZkVersion)    }
    if (!isActive) return    val partitionReplicaAssignment = zkClient.getFullReplicaAssignmentForTopics(immutable.Set(topic))    val partitionsToBeAdded = partitionReplicaAssignment.filter { case (topicPartition, _) =>      controllerContext.partitionReplicaAssignment(topicPartition).isEmpty    }
    if (topicDeletionManager.isTopicQueuedUpForDeletion(topic)) {      if (partitionsToBeAdded.nonEmpty) {        warn("Skipping adding partitions %s for topic %s since it is currently being deleted"          .format(partitionsToBeAdded.map(_._1.partition).mkString(","), topic))
        restorePartitionReplicaAssignment(topic, partitionReplicaAssignment)      } else {        // This can happen if existing partition replica assignment are restored to prevent increasing partition count during topic deletion        info("Ignoring partition change during topic deletion as no new partitions are added")      }    } else if (partitionsToBeAdded.nonEmpty) {      info(s"New partitions to be added $partitionsToBeAdded")      partitionsToBeAdded.foreach { case (topicPartition, assignedReplicas) =>        controllerContext.updatePartitionFullReplicaAssignment(topicPartition, assignedReplicas)      }      onNewPartitionCreation(partitionsToBeAdded.keySet)    }  }

  private def processPartitionReassignmentIsrChange(topicPartition: TopicPartition): Unit = {    if (!isActive) return
    if (controllerContext.partitionsBeingReassigned.contains(topicPartition)) {      val reassignment = controllerContext.partitionFullReplicaAssignment(topicPartition)      if (isReassignmentComplete(topicPartition, reassignment)) {        // resume the partition reassignment process        info(s"Target replicas ${reassignment.targetReplicas} have all caught up with the leader for " +          s"reassigning partition $topicPartition")        onPartitionReassignment(topicPartition, reassignment)      }    }  }

private def onPartitionReassignment(topicPartition: TopicPartition, reassignment: ReplicaAssignment): Unit = {    // While a reassignment is in progress, deletion is not allowed    topicDeletionManager.markTopicIneligibleForDeletion(Set(topicPartition.topic), reason = "topic reassignment in progress")
    updateCurrentReassignment(topicPartition, reassignment)
    val addingReplicas = reassignment.addingReplicas    val removingReplicas = reassignment.removingReplicas
    if (!isReassignmentComplete(topicPartition, reassignment)) {      // A1. Send LeaderAndIsr request to every replica in ORS + TRS (with the new RS, AR and RR).      updateLeaderEpochAndSendRequest(topicPartition, reassignment)      // A2. replicas in AR -> NewReplica      startNewReplicasForReassignedPartition(topicPartition, addingReplicas)    } else {      // B1. replicas in AR -> OnlineReplica      replicaStateMachine.handleStateChanges(addingReplicas.map(PartitionAndReplica(topicPartition, _)), OnlineReplica)      // B2. Set RS = TRS, AR = [], RR = [] in memory.      val completedReassignment = ReplicaAssignment(reassignment.targetReplicas)      controllerContext.updatePartitionFullReplicaAssignment(topicPartition, completedReassignment)      // B3. Send LeaderAndIsr request with a potential new leader (if current leader not in TRS) and      //   a new RS (using TRS) and same isr to every broker in ORS + TRS or TRS      moveReassignedPartitionLeaderIfRequired(topicPartition, completedReassignment)      // B4. replicas in RR -> Offline (force those replicas out of isr)      // B5. replicas in RR -> NonExistentReplica (force those replicas to be deleted)      stopRemovedReplicasOfReassignedPartition(topicPartition, removingReplicas)      // B6. Update ZK with RS = TRS, AR = [], RR = [].      updateReplicaAssignmentForPartition(topicPartition, completedReassignment)      // B7. Remove the ISR reassign listener and maybe update the /admin/reassign_partitions path in ZK to remove this partition from it.      removePartitionFromReassigningPartitions(topicPartition, completedReassignment)      // B8. After electing a leader in B3, the replicas and isr information changes, so resend the update metadata request to every broker      sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(topicPartition))      // signal delete topic thread if reassignment for some partitions belonging to topics being deleted just completed      topicDeletionManager.resumeDeletionForTopics(Set(topicPartition.topic))    }  }

  def handleAlterReplicaLogDirsRequest(request: RequestChannel.Request): Unit = {    val alterReplicaDirsRequest = request.body[AlterReplicaLogDirsRequest]    val responseMap = {      if (authorize(request, ALTER, CLUSTER, CLUSTER_NAME))        replicaManager.alterReplicaLogDirs(alterReplicaDirsRequest.partitionDirs.asScala)      else        alterReplicaDirsRequest.partitionDirs.asScala.keys.map((_, Errors.CLUSTER_AUTHORIZATION_FAILED)).toMap    }    sendResponseMaybeThrottle(request, requestThrottleMs => new AlterReplicaLogDirsResponse(requestThrottleMs, responseMap.asJava))  }

############################# Log Basics #############################
#A comma separated list of directories under which to store log fileslog.dirs=kafka-logs-5,kafka-logs-6,kafka-logs-7,kafka-logs-8

{  "version": 1,  "partitions": [{    "topic": "test_create_topic4",    "partition": 2,    "replicas": [0],    "log_dirs": ["/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-5"]  }, {    "topic": "test_create_topic4",    "partition": 1,    "replicas": [0],    "log_dirs": ["/Users/xxxxx/work/IdeaPj/source/kafka/kafka-logs-6"]  }]}