{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/37320190","id":37320190,"diff_hunk":"@@ -61,40 +61,146 @@ object AdminUtils extends Logging {\n    * p8        p9        p5        p6        p7       (2nd replica)\n    * p3        p4        p0        p1        p2       (3nd replica)\n    * p7        p8        p9        p5        p6       (3nd replica)\n+   *\n+   * To create rack aware assignment, this API will first create an rack interlaced broker list. For example,\n+   * from this brokerID -> rack mapping:\n+   *\n+   * 0 -> \"rack1\", 1 -> \"rack3\", 2 -> \"rack3\", 3 -> \"rack2\", 4 -> \"rack2\", 5 -> \"rack1\"\n+   *\n+   * The rack interlaced list will be\n+   *\n+   * 0, 3, 1, 5, 4, 2\n+   *\n+   * Then an easy round-robin assignment can be applied. Assume 6 partitions with replication factor of 3, the assignment\n+   * will be\n+   *\n+   * 0 -> 0,3,1\n+   * 1 -> 3,1,5\n+   * 2 -> 1,5,4\n+   * 3 -> 5,4,2\n+   * 4 -> 4,2,1\n+   * 5 -> 2,0,3\n+   *\n+   * Once it has completed the first round-robin, if there are more partitions to assign, the algorithm will start to\n+   * have a shift for the followers. This is to ensure we will not always get the same set of sequence.\n+   * In this case, if there is another partition to assign (partition #6), the assignment will be\n+   *\n+   * 6 -> 0,4,2 (instead of repeating 0,3,1 as partition 0)\n+   *\n+   * The rack aware assignment always chooses leader of the partition using round robin on the rack interlaced broker list.\n+   * When choosing followers, it will be biased towards brokers on racks that do not have\n+   * any replica assignment, until every rack has a replica. Then the follower assignment will go back to round-robin on\n+   * the broker list.\n+   *\n+   * As the result, if the number of replicas is equal to or greater than the number of racks, it will ensure that\n+   * each rack will get at least one replica. Otherwise, each rack will get\n+   * at most one replica. In the perfect situation where number of replica is the same as number of racks and each rack\n+   * has the same number of brokers, it guarantees that replica distribution is even across brokers and racks.\n+   *\n+   * @param rackInfo Map from broker ID to its rack (zone). If empty, no rack aware\n+   *                 assignment will be used.\n+   * @throws AdminOperationException If rack information is supplied but is incomplete, of if\n+   *                                 there is not possible to assign each replica to a unique rack\n+   *\n    */\n   def assignReplicasToBrokers(brokerList: Seq[Int],\n                               nPartitions: Int,\n                               replicationFactor: Int,\n                               fixedStartIndex: Int = -1,\n-                              startPartitionId: Int = -1)\n+                              startPartitionId: Int = -1,\n+                              rackInfo: Map[Int, String] = Map())\n   : Map[Int, Seq[Int]] = {\n+    val subRackInfo = rackInfo.filterKeys(brokerList.contains(_))\n+    if (rackInfo.size > 0 && subRackInfo.size != brokerList.size) {\n+      throw new AdminOperationException(\"Incomplete broker-rack mapping supplied for broker list \" + brokerList)\n+    }\n+    val numRacks = subRackInfo.values.toSet.size\n+    val useRackAware = numRacks > 0\n+    val arrangedBrokerList = if (useRackAware) interlaceBrokersByRack(subRackInfo) else brokerList\n     if (nPartitions <= 0)\n       throw new AdminOperationException(\"number of partitions must be larger than 0\")\n     if (replicationFactor <= 0)\n       throw new AdminOperationException(\"replication factor must be larger than 0\")\n-    if (replicationFactor > brokerList.size)\n+    if (replicationFactor > arrangedBrokerList.size)\n       throw new AdminOperationException(\"replication factor: \" + replicationFactor +\n-        \" larger than available brokers: \" + brokerList.size)\n+        \" larger than available brokers: \" + arrangedBrokerList.size)\n     val ret = new mutable.HashMap[Int, List[Int]]()\n-    val startIndex = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerList.size)\n+    val startIndex = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(arrangedBrokerList.size)\n     var currentPartitionId = if (startPartitionId >= 0) startPartitionId else 0\n \n-    var nextReplicaShift = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(brokerList.size)\n+    var nextReplicaShift = if (fixedStartIndex >= 0) fixedStartIndex else rand.nextInt(arrangedBrokerList.size)\n     for (i <- 0 until nPartitions) {\n-      if (currentPartitionId > 0 && (currentPartitionId % brokerList.size == 0))\n+      if (currentPartitionId > 0 && (currentPartitionId % arrangedBrokerList.size == 0))\n         nextReplicaShift += 1\n-      val firstReplicaIndex = (currentPartitionId + startIndex) % brokerList.size\n-      var replicaList = List(brokerList(firstReplicaIndex))\n-      for (j <- 0 until replicationFactor - 1)\n-        replicaList ::= brokerList(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerList.size))\n+      val firstReplicaIndex = (currentPartitionId + startIndex) % arrangedBrokerList.size\n+      val leader = arrangedBrokerList(firstReplicaIndex)\n+      var replicaList = List(leader)\n+      if (!useRackAware) {","path":"core/src/main/scala/kafka/admin/AdminUtils.scala","position":88,"original_position":88,"commit_id":"982c047bd826f1750934683b11f539cfbebcd677","original_commit_id":"982c047bd826f1750934683b11f539cfbebcd677","user":{"login":"hachikuji","id":12502538,"avatar_url":"https://avatars.githubusercontent.com/u/12502538?v=3","gravatar_id":"","url":"https://api.github.com/users/hachikuji","html_url":"https://github.com/hachikuji","followers_url":"https://api.github.com/users/hachikuji/followers","following_url":"https://api.github.com/users/hachikuji/following{/other_user}","gists_url":"https://api.github.com/users/hachikuji/gists{/gist_id}","starred_url":"https://api.github.com/users/hachikuji/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hachikuji/subscriptions","organizations_url":"https://api.github.com/users/hachikuji/orgs","repos_url":"https://api.github.com/users/hachikuji/repos","events_url":"https://api.github.com/users/hachikuji/events{/privacy}","received_events_url":"https://api.github.com/users/hachikuji/received_events","type":"User","site_admin":false},"body":"It might make this code a little easier to follow if you split the rack-aware and default assignments into separate functions. What do you think?","created_at":"2015-08-18T16:28:11Z","updated_at":"2015-08-18T16:28:11Z","html_url":"https://github.com/apache/kafka/pull/132#discussion_r37320190","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/132","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/37320190"},"html":{"href":"https://github.com/apache/kafka/pull/132#discussion_r37320190"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/132"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/132","id":42158890,"html_url":"https://github.com/apache/kafka/pull/132","diff_url":"https://github.com/apache/kafka/pull/132.diff","patch_url":"https://github.com/apache/kafka/pull/132.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/132","number":132,"state":"open","locked":false,"title":"KAFKA-1215: Rack-Aware replica assignment option","user":{"login":"allenxwang","id":1728105,"avatar_url":"https://avatars.githubusercontent.com/u/1728105?v=3","gravatar_id":"","url":"https://api.github.com/users/allenxwang","html_url":"https://github.com/allenxwang","followers_url":"https://api.github.com/users/allenxwang/followers","following_url":"https://api.github.com/users/allenxwang/following{/other_user}","gists_url":"https://api.github.com/users/allenxwang/gists{/gist_id}","starred_url":"https://api.github.com/users/allenxwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allenxwang/subscriptions","organizations_url":"https://api.github.com/users/allenxwang/orgs","repos_url":"https://api.github.com/users/allenxwang/repos","events_url":"https://api.github.com/users/allenxwang/events{/privacy}","received_events_url":"https://api.github.com/users/allenxwang/received_events","type":"User","site_admin":false},"body":"The PR tries to achieve the following:\r\n\r\n- Make rack-aware assignment and rack data structure optional as opposed to be part of the core data structure/protocol to ease the migration. The implementation of that returns the map of broker to rack is pluggable. User needs to pass the implementation class as a Kafka runtime configuration or command line argument.\r\n\r\n- The rack aware replica assignment is best effort when distributing the replicas to racks. When there are more replicas than racks, it ensures each rack has at least one replica. When there are more racks than replicas, it ensures each rack has at most one replica. It also tries to keep the even distribution of replicas among brokers and racks when possible.\r\n","created_at":"2015-08-11T18:06:08Z","updated_at":"2015-08-18T16:28:11Z","closed_at":null,"merged_at":null,"merge_commit_sha":"a5d86452efaaf0e0e0b7906f1e2351597ce8d821","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/132/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/132/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/132/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/982c047bd826f1750934683b11f539cfbebcd677","head":{"label":"allenxwang:KAFKA-1215","ref":"KAFKA-1215","sha":"982c047bd826f1750934683b11f539cfbebcd677","user":{"login":"allenxwang","id":1728105,"avatar_url":"https://avatars.githubusercontent.com/u/1728105?v=3","gravatar_id":"","url":"https://api.github.com/users/allenxwang","html_url":"https://github.com/allenxwang","followers_url":"https://api.github.com/users/allenxwang/followers","following_url":"https://api.github.com/users/allenxwang/following{/other_user}","gists_url":"https://api.github.com/users/allenxwang/gists{/gist_id}","starred_url":"https://api.github.com/users/allenxwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allenxwang/subscriptions","organizations_url":"https://api.github.com/users/allenxwang/orgs","repos_url":"https://api.github.com/users/allenxwang/repos","events_url":"https://api.github.com/users/allenxwang/events{/privacy}","received_events_url":"https://api.github.com/users/allenxwang/received_events","type":"User","site_admin":false},"repo":{"id":25887909,"name":"kafka","full_name":"allenxwang/kafka","owner":{"login":"allenxwang","id":1728105,"avatar_url":"https://avatars.githubusercontent.com/u/1728105?v=3","gravatar_id":"","url":"https://api.github.com/users/allenxwang","html_url":"https://github.com/allenxwang","followers_url":"https://api.github.com/users/allenxwang/followers","following_url":"https://api.github.com/users/allenxwang/following{/other_user}","gists_url":"https://api.github.com/users/allenxwang/gists{/gist_id}","starred_url":"https://api.github.com/users/allenxwang/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allenxwang/subscriptions","organizations_url":"https://api.github.com/users/allenxwang/orgs","repos_url":"https://api.github.com/users/allenxwang/repos","events_url":"https://api.github.com/users/allenxwang/events{/privacy}","received_events_url":"https://api.github.com/users/allenxwang/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/allenxwang/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/allenxwang/kafka","forks_url":"https://api.github.com/repos/allenxwang/kafka/forks","keys_url":"https://api.github.com/repos/allenxwang/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/allenxwang/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/allenxwang/kafka/teams","hooks_url":"https://api.github.com/repos/allenxwang/kafka/hooks","issue_events_url":"https://api.github.com/repos/allenxwang/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/allenxwang/kafka/events","assignees_url":"https://api.github.com/repos/allenxwang/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/allenxwang/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/allenxwang/kafka/tags","blobs_url":"https://api.github.com/repos/allenxwang/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/allenxwang/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/allenxwang/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/allenxwang/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/allenxwang/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/allenxwang/kafka/languages","stargazers_url":"https://api.github.com/repos/allenxwang/kafka/stargazers","contributors_url":"https://api.github.com/repos/allenxwang/kafka/contributors","subscribers_url":"https://api.github.com/repos/allenxwang/kafka/subscribers","subscription_url":"https://api.github.com/repos/allenxwang/kafka/subscription","commits_url":"https://api.github.com/repos/allenxwang/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/allenxwang/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/allenxwang/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/allenxwang/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/allenxwang/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/allenxwang/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/allenxwang/kafka/merges","archive_url":"https://api.github.com/repos/allenxwang/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/allenxwang/kafka/downloads","issues_url":"https://api.github.com/repos/allenxwang/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/allenxwang/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/allenxwang/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/allenxwang/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/allenxwang/kafka/labels{/name}","releases_url":"https://api.github.com/repos/allenxwang/kafka/releases{/id}","created_at":"2014-10-28T20:08:20Z","updated_at":"2014-10-28T07:04:48Z","pushed_at":"2015-08-14T22:22:04Z","git_url":"git://github.com/allenxwang/kafka.git","ssh_url":"git@github.com:allenxwang/kafka.git","clone_url":"https://github.com/allenxwang/kafka.git","svn_url":"https://github.com/allenxwang/kafka","homepage":null,"size":21051,"stargazers_count":0,"watchers_count":0,"language":null,"has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"0.8.1"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"e2ebae80982c45c65f709efaedd89d10e109600f","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"name":"kafka","full_name":"apache/kafka","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","created_at":"2011-08-15T18:06:16Z","updated_at":"2015-08-18T16:24:22Z","pushed_at":"2015-08-18T16:10:41Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":30805,"stargazers_count":1811,"watchers_count":1811,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1074,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":61,"forks":1074,"open_issues":61,"watchers":1811,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/132"},"html":{"href":"https://github.com/apache/kafka/pull/132"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/132"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/132/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/132/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/132/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/982c047bd826f1750934683b11f539cfbebcd677"}}}}