defhandleCreatePartitionsRequest(request: RequestChannel.Request): Unit = { val createPartitionsRequest = request.body[CreatePartitionsRequest]
//部分代码省略..
//如果当前不是Controller角色直接抛出异常 if (!controller.isActive) { val result = createPartitionsRequest.data.topics.asScala.map { topic => (topic.name, newApiError(Errors.NOT_CONTROLLER, null)) }.toMap sendResponseCallback(result) } else { // Special handling to add duplicate topics to the response val topics = createPartitionsRequest.data.topics.asScala val dupes = topics.groupBy(_.name) .filter { _._2.size > 1 } .keySet val notDuped = topics.filterNot(topic => dupes.contains(topic.name)) val authorizedTopics = filterAuthorized(request, ALTER, TOPIC, notDuped.map(_.name)) val (authorized, unauthorized) = notDuped.partition { topic => authorizedTopics.contains(topic.name) }
val (queuedForDeletion, valid) = authorized.partition { topic => controller.topicDeletionManager.isTopicQueuedUpForDeletion(topic.name) }
val errors = dupes.map(_ -> newApiError(Errors.INVALID_REQUEST, "Duplicate topic in request.")) ++ unauthorized.map(_.name -> newApiError(Errors.TOPIC_AUTHORIZATION_FAILED, "The topic authorization is failed.")) ++ queuedForDeletion.map(_.name -> newApiError(Errors.INVALID_TOPIC_EXCEPTION, "The topic is queued for deletion."))
调用 adminManager.createPartitions 3.1 从zk中获取/brokers/ids/Brokers列表的元信息的 3.2 从zk获取/brokers/topics/{topicName}已经存在的副本分配方式,并判断是否有正在进行副本重分配的进程在执行,如果有的话就抛出异常结束流程 3.3 如果从zk获取/brokers/topics/{topicName}数据不存在则抛出异常 The topic '$topic' does not exist 3.4 检查修改的分区数是否比原来的分区数大，如果比原来还小或者等于原来分区数则抛出异常结束流程 3.5 如果传入的参数--replica-assignment 中有不存在的BrokerId;则抛出异常Unknown broker(s) in replica assignment结束流程 3.5 如果传入的--partitions数量 与--replica-assignment中新增的部分数量不匹配则抛出异常Increasing the number of partitions by... 结束流程 3.6 调用 adminZkClient.addPartitions
adminZkClient.addPartitions 添加分区
校验--partitions数量是否比存在的分区数大,否则异常The number of partitions for a topic can only be increased
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") } } elseif (partitionsToBeAdded.nonEmpty) { info(s"New partitions to be added $partitionsToBeAdded") partitionsToBeAdded.foreach { case (topicPartition, assignedReplicas) => controllerContext.updatePartitionFullReplicaAssignment(topicPartition, assignedReplicas) } onNewPartitionCreation(partitionsToBeAdded.keySet) } }
/** * This callback is invoked by the topic change callback with the list of failed brokers as input. * It does the following - * 1. Move the newly created partitions to the NewPartition state * 2. Move the newly created partitions from NewPartition->OnlinePartition state */ privatedefonNewPartitionCreation(newPartitions: Set[TopicPartition]): Unit = { info(s"New partition creation callback for ${newPartitions.mkString(",")}") partitionStateMachine.handleStateChanges(newPartitions.toSeq, NewPartition) replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions).toSeq, NewReplica) partitionStateMachine.handleStateChanges( newPartitions.toSeq, OnlinePartition, Some(OfflinePartitionLeaderElectionStrategy(false)) ) replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions).toSeq, OnlineReplica) }
/** * 如果日志已经存在，只返回现有日志的副本否则如果 isNew=true 或者如果没有离线日志目录，则为给定的主题和给定的分区创建日志 否则抛出 KafkaStorageException */ defgetOrCreateLog(topicPartition: TopicPartition, config: LogConfig, isNew: Boolean = false, isFuture: Boolean = false): Log = { logCreationOrDeletionLock synchronized { getLog(topicPartition, isFuture).getOrElse { // create the log if it has not already been created in another thread if (!isNew && offlineLogDirs.nonEmpty) thrownewKafkaStorageException(s"Can not create log for $topicPartition because log directories ${offlineLogDirs.mkString(",")} are offline")
val logDirs: List[File] = { val preferredLogDir = preferredLogDirs.get(topicPartition)
if (isFuture) { if (preferredLogDir == null) thrownewIllegalStateException(s"Can not create the future log for $topicPartition without having a preferred log directory") elseif (getLog(topicPartition).get.dir.getParent == preferredLogDir) thrownewIllegalStateException(s"Can not create the future log for $topicPartition in the current log directory of this partition") }
if (preferredLogDir != null) List(newFile(preferredLogDir)) else nextLogDirs() }
val logDirName = { if (isFuture) Log.logFutureDirName(topicPartition) else Log.logDirName(topicPartition) }
val logDir = logDirs .toStream // to prevent actually mapping the whole list, lazy map .map(createLogDirectory(_, logDirName)) .find(_.isSuccess) .getOrElse(Failure(newKafkaStorageException("No log directories available. Tried " + logDirs.map(_.getAbsolutePath).mkString(", ")))) .get // If Failure, will throw
val log = Log( dir = logDir, config = config, logStartOffset = 0L, recoveryPoint = 0L, maxProducerIdExpirationMs = maxPidExpirationMs, producerIdExpirationCheckIntervalMs = LogManager.ProducerIdExpirationCheckIntervalMs, scheduler = scheduler, time = time, brokerTopicStats = brokerTopicStats, logDirFailureChannel = logDirFailureChannel)
if (isFuture) futureLogs.put(topicPartition, log) else currentLogs.put(topicPartition, log)
info(s"Created log for partition $topicPartition in $logDir with properties " + s"{${config.originals.asScala.mkString(", ")}}.") // Remove the preferred log dir since it has already been satisfied preferredLogDirs.remove(topicPartition)
微信
