一、命令创建topic

kafka-topics --create --topic quickstart-events --bootstrap-server cdh1:9092 --partitions 2 --replication-factor 2

二、kafka-topics脚本

exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.tools.TopicCommand "$@"

脚本中指定了处理它的主类：TopicCommand

三、TopicCommand

public abstract class TopicCommand {public static void main(String... args) {Exit.exit(mainNoExit(args));}private static int mainNoExit(String... args) {try {execute(args);return 0;} catch (Throwable e) {return 1;}}static void execute(String... args) throws Exception {//解析命令行参数TopicCommandOptions opts = new TopicCommandOptions(args);//创建TopicServiceTopicService topicService = new TopicService(opts.commandConfig(), opts.bootstrapServer());try {if (opts.hasCreateOption()) {//这是处理topic创建的，我们主要分析它topicService.createTopic(opts);} else if (opts.hasAlterOption()) {//更高topic逻辑topicService.alterTopic(opts);} else if (opts.hasListOption()) {//获取topictopicService.listTopics(opts);} else if (opts.hasDescribeOption()) {//topi相关描述信息topicService.describeTopic(opts);} else if (opts.hasDeleteOption()) {//删除topictopicService.deleteTopic(opts);}}catch(...){...}finally {topicService.close();}}public static class TopicService implements AutoCloseable {public void createTopic(TopicCommandOptions opts) throws Exception {CommandTopicPartition topic = new CommandTopicPartition(opts);if (Topic.hasCollisionChars(topic.name)) {//由于度量名称的限制，带有句点（“.”）或下划线（“_”）的主题可能会发生冲突。为了避免问题，最好使用其中之一，但不要两者都使用System.out.println(".........");}createTopic(topic);}public void createTopic(CommandTopicPartition topic) throws Exception {if (topic.replicationFactor.filter(rf -> rf > Short.MAX_VALUE || rf < 1).isPresent()) {//复制因子必须介于1和“+Short.MAX_VALUE+”之间throw new IllegalArgumentException("...");}if (topic.partitions.filter(p -> p < 1).isPresent()) {//分区必须大于0throw new IllegalArgumentException("...");}try {NewTopic newTopic;//取决于创建 topic 时 是否指定了   replica-assignmentif (topic.hasReplicaAssignment()) {newTopic = new NewTopic(topic.name, topic.replicaAssignment);} else {newTopic = new NewTopic(topic.name, topic.partitions, topic.replicationFactor.map(Integer::shortValue));}//给topic设置参数Map<String, String> configsMap = topic.configsToAdd.stringPropertyNames().stream().collect(Collectors.toMap(name -> name, name -> topic.configsToAdd.getProperty(name)));newTopic.configs(configsMap);//批量创建topicCreateTopicsResult createResult = adminClient.createTopics(Collections.singleton(newTopic),new CreateTopicsOptions().retryOnQuotaViolation(false));//等待所有topic都创建成功createResult.all().get();System.out.println("Created topic " + topic.name + ".");} catch (ExecutionException e) {//......}}}}

TopicCommandOptions中有对创建topic所有参数的解读，我们下面来详细看下这些参数

四、创建Topic参数

bootstrap-server

        必选项：连接Kafka server用

command-config

        包含要传递给Admin Client的配置的属性文件。这仅与--bootstrap-server选项一起使用，用于描述和更改broker配置

list

        列出所有可用的topic

create

        创建一个新的topic

delete

        删除一个topic

alter

        更改分区数量和副本分配，通过--alter更新现有主题的配置

describe

        列出给定topic的详细信息

topic

        要创建、更改、描述或删除的主题。它还接受正则表达式，但--create选项除外。将主题名称放在双引号中，并使用“\\”前缀转义正则表达式符号；例如 \"test\\.topic\"

topic-id

        仅与用于描述主题的--bootstrap-server选项一起使用

config

        正在创建的主题的主题配置覆盖。

delete-config

        要删除现有主题的主题配置覆盖

partitions

        正在创建或更改的主题的分区数量（警告：如果为具有键的主题增加分区，则分区逻辑或消息顺序将受到影响）。如果未提供用于，则为集群默认值

replication-factor

        正在创建的主题中每个分区的复制因子。如果未提供，则为群集默认值

replica-assignment

        正在创建或更改的topic的手动分区到broker分配列表

under-replicated-partitions

        如果在描述主题时设置，则仅在复制分区下显示

unavailable-partitions

        如果在描述主题时设置，则仅显示其leader不可用的分区

under-min-isr-partitions

        如果在描述主题时设置，则仅显示isr计数 < 配置的最小值的分区。

at-min-isr-partitions

        如果在描述主题时设置，则仅显示isr计数 = 配置的最小值的分区        

topics-with-overrides

        如果在描述主题时设置，则仅显示已覆盖配置的topic

if-exists

        如果在更改、删除或描述主题时设置，则仅当主题存在时才会执行该操作

if-not-exists

        如果在创建主题时设置，则仅当主题不存在时才会执行该操作。

exclude-internal

        运行list或describe命令时排除内部topic。默认情况下，内部topic将被列出

partition-size-limit-per-response

        一个DescribeTopicPartitions响应中包含的最大分区大小

五、AdminClient

从第二步的源码中看到最终将topic的创建交给了AdminClient来完成，下面我们继续往下分析

1、创建

在TopicService的构造方法中创建的AdminClient

它是Kafka的管理客户端，支持管理和检查topic、broker、配置和ACL。

AdminClient的创建用到了bootstrap.servers，它里面有连接KafkaServer的host:port列表。

bootstrap.servers配置仅用于发现群集中的broker，然后AdminClient将根据需要连接到这些broker。因此，只包括两个或三个经纪人地址就足以应对broker不可用的风险。

TopicService topicService = new TopicService(opts.commandConfig(), opts.bootstrapServer());

    public static class TopicService implements AutoCloseable {private final Admin adminClient;public TopicService(Properties commandConfig, Optional<String> bootstrapServer) {this.adminClient = createAdminClient(commandConfig, bootstrapServer);}private static Admin createAdminClient(Properties commandConfig, Optional<String> bootstrapServer) {if (bootstrapServer.isPresent()) {commandConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer.get());}return Admin.create(commandConfig);}}

2、交由子类KafkaAdminClient处理

public class KafkaAdminClient extends AdminClient {private final AdminClientRunnable runnable;//创建一批topicpublic CreateTopicsResult createTopics(final Collection<NewTopic> newTopics,final CreateTopicsOptions options) {final Map<String, KafkaFutureImpl<TopicMetadataAndConfig>> topicFutures = new HashMap<>(newTopics.size());final CreatableTopicCollection topics = new CreatableTopicCollection();for (NewTopic newTopic : newTopics) {//判断名字是否符合规范if (topicNameIsUnrepresentable(newTopic.name())) {KafkaFutureImpl<TopicMetadataAndConfig> future = new KafkaFutureImpl<>();future.completeExceptionally(new InvalidTopicException("The given topic name '" +newTopic.name() + "' cannot be represented in a request."));topicFutures.put(newTopic.name(), future);} else if (!topicFutures.containsKey(newTopic.name())) {//topicFutures 装的是还没有创建的 topicnametopicFutures.put(newTopic.name(), new KafkaFutureImpl<>());topics.add(newTopic.convertToCreatableTopic());}}if (!topics.isEmpty()) {final long now = time.milliseconds();final long deadline = calcDeadlineMs(now, options.timeoutMs());//里面封装了 ApiKeys.CREATE_TOPICS 请求final Call call = getCreateTopicsCall(options, topicFutures, topics,Collections.emptyMap(), now, deadline);//实现了Runnable接口runnable.call(call, now);}return new CreateTopicsResult(new HashMap<>(topicFutures));}
}

从这里我们看到，这里会用一个线程向broker发送ApiKeys.CREATE_TOPICS 请求。下面我们来看broker端怎么处理topics的创建请求的。按照我们之前的经验，要去看KafkaApis中对应ApiKeys.CREATE_TOPICS的处理逻辑

class KafkaApis(...）{request.header.apiKey match {//....case ApiKeys.CREATE_TOPICS => maybeForwardToController(request, handleCreateTopicsRequest)case ApiKeys.DELETE_TOPICS => maybeForwardToController(request, handleDeleteTopicsRequest)case ApiKeys.CREATE_ACLS => maybeForwardToController(request, handleCreateAcls)case ApiKeys.DELETE_ACLS => maybeForwardToController(request, handleDeleteAcls)case ApiKeys.CREATE_PARTITIONS => maybeForwardToController(request, handleCreatePartitionsRequest)case ApiKeys.ELECT_LEADERS => maybeForwardToController(request, handleElectLeaders)//.....}
}

六、CREATE_TOPICS的处理逻辑

从KafkaApi中我们看到很多请求都调用了maybeForwardToController()方法来处理，但是传入的参数不同，从名称上我们可以猜测这些请求可能交由Controller来处理，回想下《Kafka-Controller角色需要做什么？》中当一个broker当选为Controller时第一件事就是注册监听器，去监听broker改变、topic改变、topic删除、isr改变等，并分别准备好了响应的处理逻辑。因此这里只要让topic发生改变就可以自动触发让Controller处理了。下面看下handleCreateTopicsRequest()中都做了什么？

1、获取ZooKeeper

val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldAlwaysForward(request))

2、判断集群当下是否有Controller

如果集群当下没有Controller，直接向客户端返回Errors.NOT_CONTROLLER错误。我们按照集群当下有Controller继续分析。

    if (!zkSupport.controller.isActive) {//如果没有contorller，直接向客户端发送响应信息（集群当下没有controller），且这个时候时创建不了topic的，createTopicsRequest.data.topics.forEach { topic =>results.add(new CreatableTopicResult().setName(topic.name).setErrorCode(Errors.NOT_CONTROLLER.code))}sendResponseCallback(results)} else {//正常逻辑}

3、检查topic名称

集群元数据topic是一个具有不同实现的内部topic。不应允许用户创建同名的topic。

          if (topicNames.contains(Topic.CLUSTER_METADATA_TOPIC_NAME)) {//拒绝创建内部主题  __cluster_metadatainfo(s"Rejecting creation of internal topic ${Topic.CLUSTER_METADATA_TOPIC_NAME}")}topicNames.diff(Set(Topic.CLUSTER_METADATA_TOPIC_NAME))

4、调用ZkAdminManager创建topic

      zkSupport.adminManager.createTopics(createTopicsRequest.data.timeoutMs,createTopicsRequest.data.validateOnly,toCreate,authorizedForDescribeConfigs,controllerMutationQuota,handleCreateTopicsResults)}

1、循环校验每个topic是否符合规则

1、topic是否已经存在

2、topic是否为null

3、numPartitions或replicationFactor和replicasAssignments都已设置。两者不能同时使用

2、确定分区分配列表

如果用户指定了列表，那么就直接用用户的，否则使用Kafka自己的分配策略（下篇博客分析）

        val assignments = if (topic.assignments.isEmpty) {CoreUtils.replicaToBrokerAssignmentAsScala(AdminUtils.assignReplicasToBrokers(brokers.asJavaCollection, resolvedNumPartitions, resolvedReplicationFactor))} else {val assignments = new mutable.HashMap[Int, Seq[Int]]//注意：我们不会检查replicaAssignment是否包含未知的代理——与添加分区的情况不同，这遵循TopicCommand中的现有逻辑topic.assignments.forEach { assignment =>assignments(assignment.partitionIndex) = assignment.brokerIds.asScala.map(a => a: Int)}assignments}

3、topics目录下创建指定的topic

//ConfigType.TOPIC : topics 目录
//topic ：要创建的topic名称
zkClient.setOrCreateEntityConfigs(ConfigType.TOPIC, topic, config)

4、topic目录下创建分区目录和对应信息

    writeTopicPartitionAssignment(topic, partitionReplicaAssignment.map { case (k, v) => k -> ReplicaAssignment(v) },isUpdate = false, usesTopicId)

5、创建对应的元数据

CreatePartitionsMetadata(topic.name, assignments.keySet)

七、Controller端处理逻辑

我们找到TopicChange对应的处理逻辑

  override def process(event: ControllerEvent): Unit = {try {event match {case TopicChange =>processTopicChange()//......}}}private def processTopicChange(): Unit = {if (!isActive) return//从 brokers/topics/目录下获取所有的topicval topics = zkClient.getAllTopicsInCluster(true)//从controllerContext 获取当下缓存中所有的 topic//两者相减获取 新增加的 topicval newTopics = topics -- controllerContext.allTopics// 获取删除的topic （既topics目录没有，但是缓存中有）val deletedTopics = controllerContext.allTopics.diff(topics)//设置新的topic到缓存controllerContext.setAllTopics(topics)//检测zk中 每个topic 目录的变化registerPartitionModificationsHandlers(newTopics.toSeq)//现在要添加分区和副本了，也就是从topic下获取 topic_id、adding_replicas、removing_replicas、partitions 信息val addedPartitionReplicaAssignment = zkClient.getReplicaAssignmentAndTopicIdForTopics(newTopics)deletedTopics.foreach(controllerContext.removeTopic)processTopicIds(addedPartitionReplicaAssignment)addedPartitionReplicaAssignment.foreach { case TopicIdReplicaAssignment(_, _, newAssignments) =>newAssignments.foreach { case (topicAndPartition, newReplicaAssignment) =>//controllerContext 的缓存中 更新分区、副本、leder信息controllerContext.updatePartitionFullReplicaAssignment(topicAndPartition, newReplicaAssignment)}}info(s"New topics: [$newTopics], deleted topics: [$deletedTopics], new partition replica assignment " +s"[$addedPartitionReplicaAssignment]")if (addedPartitionReplicaAssignment.nonEmpty) {val partitionAssignments = addedPartitionReplicaAssignment.map { case TopicIdReplicaAssignment(_, _, partitionsReplicas) => partitionsReplicas.keySet }.reduce((s1, s2) => s1.union(s2))//更高topic下的分区、副本为可用状态 OnlineReplica//此时 往topic 生产数据就ok了 onNewPartitionCreation(partitionAssignments)}}

 从源码中我们可以看到，Controller这端会不断的将新的topic以及其下的topic_id、adding_replicas、removing_replicas、partitions 信息加载到缓存，并使用它们的状态机将它们更新至可用状态。并剔除掉删除的topic。始终保持，当向topic生产数据时，它这里都时最新的状态。