赞
踩
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 {
val assignments = if (topic.assignments().isEmpty) {
AdminUtils.assignReplicasToBrokers(
brokers, resolvedNumPartitions, resolvedReplicationFactor)
} else {
val assignments = new mutable.HashMap[Int, Seq[Int]]
// Note: we don’t check that replicaAssignment contains unknown brokers - unlike in add-partitions case,
// this follows the existing logic in TopicCommand
topic.assignments.asScala.foreach {
case assignment => assignments(assignment.partitionIndex()) =
assignment.brokerIds().asScala.map(a => a: Int)
}
assignments
}
trace(s"Assignments for topic $topic are $assignments ")
}
--replica-assignment
;一种是自己指定了分区分配从源码中得知, 会把我们指定的规则进行了包装,注意它并没有去检查你指定的Broker是否存在;
参数检查: 分区数>0; 副本数>0; 副本数<=Broker数 (如果自己未定义会直接使用Broker中个配置)
根据是否有 机架信息来进行不同方式的分配;
要么整个集群都有机架信息,要么整个集群都没有机架信息; 否则抛出异常
副本分配的几个原则:
将副本平均分布在所有的 Broker 上;
partition 的多个副本应该分配在不同的 Broker 上;
如果所有的 Broker 有机架信息的话, partition 的副本应该分配到不同的机架上。
AdminUtils.assignReplicasToBrokersRackUnaware<
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。