你知道 Kafka 创建 Topic 这个过程做了哪些事情吗？(附视频)

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic test

object AdminClientTopicService {    def createAdminClient(commandConfig: Properties, bootstrapServer: Option[String]): Admin = {      bootstrapServer match {        case Some(serverList) => commandConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, serverList)        case None =>      }      Admin.create(commandConfig)    }​    def apply(commandConfig: Properties, bootstrapServer: Option[String]): AdminClientTopicService =      new AdminClientTopicService(createAdminClient(commandConfig, bootstrapServer))  }

  case class AdminClientTopicService private (adminClient: Admin) extends TopicService {​    override def createTopic(topic: CommandTopicPartition): Unit = {      //如果配置了副本副本数--replication-factor 一定要大于0      if (topic.replicationFactor.exists(rf => rf > Short.MaxValue || rf < 1))        throw new IllegalArgumentException(s"The replication factor must be between 1 and ${Short.MaxValue} inclusive")       //如果配置了--partitions 分区数 必须大于0      if (topic.partitions.exists(partitions => partitions < 1))        throw new IllegalArgumentException(s"The partitions must be greater than 0")​    //查询是否已经存在该Topic      if (!adminClient.listTopics().names().get().contains(topic.name)) {        val newTopic = if (topic.hasReplicaAssignment)          //如果指定了--replica-assignment参数；则按照指定的来分配副本          new NewTopic(topic.name, asJavaReplicaReassignment(topic.replicaAssignment.get))        else {          new NewTopic(            topic.name,            topic.partitions.asJava,            topic.replicationFactor.map(_.toShort).map(Short.box).asJava)        }​        // 将配置--config 解析成一个配置map        val configsMap = topic.configsToAdd.stringPropertyNames()          .asScala          .map(name => name -> topic.configsToAdd.getProperty(name))          .toMap.asJava​        newTopic.configs(configsMap)        //调用adminClient创建Topic        val createResult = adminClient.createTopics(Collections.singleton(newTopic))        createResult.all().get()        println(s"Created topic ${topic.name}.")      } else {        throw new IllegalArgumentException(s"Topic ${topic.name} already exists")      }    }

    @Override    public CreateTopicsResult createTopics(final Collection<NewTopic> newTopics,                                           final CreateTopicsOptions options) {              //省略部分源码...        Call call = new Call("createTopics", calcDeadlineMs(now, options.timeoutMs()),            new ControllerNodeProvider()) {​            @Override            public CreateTopicsRequest.Builder createRequest(int timeoutMs) {                return new CreateTopicsRequest.Builder(                    new CreateTopicsRequestData().                        setTopics(topics).                        setTimeoutMs(timeoutMs).                        setValidateOnly(options.shouldValidateOnly()));            }​            @Override            public void handleResponse(AbstractResponse abstractResponse) {                //省略            }​            @Override            void handleFailure(Throwable throwable) {                completeAllExceptionally(topicFutures.values(), throwable);            }        };            }

​def handleCreateTopicsRequest(request: RequestChannel.Request): Unit = {    // 部分代码省略  //如果当前Broker不是属于Controller的话,就抛出异常    if (!controller.isActive) {      createTopicsRequest.data.topics.asScala.foreach { topic =>        results.add(new CreatableTopicResult().setName(topic.name).          setErrorCode(Errors.NOT_CONTROLLER.code))      }      sendResponseCallback(results)    } else {     // 部分代码省略    }      adminManager.createTopics(createTopicsRequest.data.timeoutMs,          createTopicsRequest.data.validateOnly,          toCreate,          authorizedForDescribeConfigs,          handleCreateTopicsResults)    }  }​

def createTopics(timeout: Int,                   validateOnly: Boolean,                   toCreate: Map[String, CreatableTopic],                   includeConfigsAndMetatadata: Map[String, CreatableTopicResult],                   responseCallback: Map[String, ApiError] => Unit): Unit = {​    // 1. map over topics creating assignment and calling zookeeper    val brokers = metadataCache.getAliveBrokers.map { b => kafka.admin.BrokerMetadata(b.id, b.rack) }    val metadata = toCreate.values.map(topic =>      try {          //省略部分代码         //检查Topic是否存在         //检查 --replica-assignment参数和 (--partitions || --replication-factor ) 不能同时使用         // 如果(--partitions || --replication-factor ) 没有设置,则使用 Broker的配置(这个Broker肯定是Controller)    // 计算分区副本分配方式​        createTopicPolicy match {          case Some(policy) =>          //省略部分代码            adminZkClient.validateTopicCreate(topic.name(), assignments, configs)            if (!validateOnly)              adminZkClient.createTopicWithAssignment(topic.name, configs, assignments)​          case None =>            if (validateOnly)             //校验创建topic的参数准确性              adminZkClient.validateTopicCreate(topic.name, assignments, configs)            else              //把topic相关数据写入到zk中              adminZkClient.createTopicWithAssignment(topic.name, configs, assignments)        }​         }

  def createTopicWithAssignment(topic: String,                                config: Properties,                                partitionReplicaAssignment: Map[Int, Seq[Int]]): Unit = {    validateTopicCreate(topic, partitionReplicaAssignment, config)​    // 将topic单独的配置写入到zk中    zkClient.setOrCreateEntityConfigs(ConfigType.Topic, topic, config)​    // 将topic分区相关信息写入zk中    writeTopicPartitionAssignment(topic, partitionReplicaAssignment.mapValues(ReplicaAssignment(_)).toMap, isUpdate = false)  }​

​  private def processTopicChange(): Unit = {    //如果处理的不是Controller角色就返回    if (!isActive) return    //从zk中获取 `/brokers/topics 所有Topic    val topics = zkClient.getAllTopicsInCluster    //找出哪些是新增的    val newTopics = topics -- controllerContext.allTopics    //找出哪些Topic在zk上被删除了    val deletedTopics = controllerContext.allTopics -- topics    controllerContext.allTopics = topics​        registerPartitionModificationsHandlers(newTopics.toSeq)    val addedPartitionReplicaAssignment = zkClient.getFullReplicaAssignmentForTopics(newTopics)    deletedTopics.foreach(controllerContext.removeTopic)    addedPartitionReplicaAssignment.foreach {      case (topicAndPartition, newReplicaAssignment) => controllerContext.updatePartitionFullReplicaAssignment(topicAndPartition, newReplicaAssignment)    }    info(s"New topics: [$newTopics], deleted topics: [$deletedTopics], new partition replica assignment " +      s"[$addedPartitionReplicaAssignment]")    if (addedPartitionReplicaAssignment.nonEmpty)      onNewPartitionCreation(addedPartitionReplicaAssignment.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   */  private def onNewPartitionCreation(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   */  def getOrCreateLog(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)          throw new KafkaStorageException(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)              throw new IllegalStateException(s"Can not create the future log for $topicPartition without having a preferred log directory")            else if (getLog(topicPartition).get.dir.getParent == preferredLogDir)              throw new IllegalStateException(s"Can not create the future log for $topicPartition in the current log directory of this partition")          }​          if (preferredLogDir != null)            List(new File(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(new KafkaStorageException("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)​        log      }    }  }

configurations:                                        cleanup.policy                                          compression.type                                        delete.retention.ms                                     file.delete.delay.ms                                    flush.messages                                          flush.ms                                                follower.replication.throttled.       replicas                                             index.interval.bytes                                  leader.replication.throttled.replicas                 max.compaction.lag.ms                                 max.message.bytes                                     message.downconversion.enable                         message.format.version                                message.timestamp.difference.max.ms                   message.timestamp.type                                min.cleanable.dirty.ratio                             min.compaction.lag.ms                                 min.insync.replicas                                   preallocate                                           retention.bytes                                       retention.ms                                          segment.bytes                                         segment.index.bytes                                   segment.jitter.ms                                     segment.ms                                            unclean.leader.election.enable