{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/151766517","pull_request_review_id":77531883,"id":151766517,"diff_hunk":"@@ -0,0 +1,417 @@\n+/**\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements.  See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License.  You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+*/\n+package kafka.zk\n+\n+import java.util.Properties\n+\n+import kafka.admin.{AdminOperationException, AdminUtils, BrokerMetadata, RackAwareMode}\n+import kafka.common.TopicAlreadyMarkedForDeletionException\n+import kafka.log.LogConfig\n+import kafka.server.{ConfigEntityName, ConfigType, DynamicConfig}\n+import kafka.utils._\n+import org.apache.kafka.common.TopicPartition\n+import org.apache.kafka.common.errors._\n+import org.apache.kafka.common.internals.Topic\n+import org.apache.zookeeper.KeeperException.NodeExistsException\n+\n+import scala.collection.{Map, Seq}\n+\n+class AdminZkClient(zkClient: KafkaZkClient) extends Logging {\n+\n+  /**\n+   * Creates the topic with given configuration\n+   * @param topic topic name to create\n+   * @param partitions  Number of partitions to be set\n+   * @param replicationFactor Replication factor\n+   * @param topicConfig  topic configs\n+   * @param rackAwareMode\n+   */\n+  def createTopic(topic: String,\n+                  partitions: Int,\n+                  replicationFactor: Int,\n+                  topicConfig: Properties = new Properties,\n+                  rackAwareMode: RackAwareMode = RackAwareMode.Enforced) {\n+    val brokerMetadatas = getBrokerMetadatas(rackAwareMode)\n+    val replicaAssignment = AdminUtils.assignReplicasToBrokers(brokerMetadatas, partitions, replicationFactor)\n+    createOrUpdateTopicPartitionAssignmentPathInZK(topic, replicaAssignment, topicConfig)\n+  }\n+\n+  /**\n+   * Gets broker metadata list\n+   * @param rackAwareMode\n+   * @param brokerList\n+   * @return\n+   */\n+  def getBrokerMetadatas(rackAwareMode: RackAwareMode = RackAwareMode.Enforced,\n+                         brokerList: Option[Seq[Int]] = None): Seq[BrokerMetadata] = {\n+    val allBrokers = zkClient.getAllBrokersInCluster\n+    val brokers = brokerList.map(brokerIds => allBrokers.filter(b => brokerIds.contains(b.id))).getOrElse(allBrokers)\n+    val brokersWithRack = brokers.filter(_.rack.nonEmpty)\n+    if (rackAwareMode == RackAwareMode.Enforced && brokersWithRack.nonEmpty && brokersWithRack.size < brokers.size) {\n+      throw new AdminOperationException(\"Not all brokers have rack information. Add --disable-rack-aware in command line\" +\n+        \" to make replica assignment without rack information.\")\n+    }\n+    val brokerMetadatas = rackAwareMode match {\n+      case RackAwareMode.Disabled => brokers.map(broker => BrokerMetadata(broker.id, None))\n+      case RackAwareMode.Safe if brokersWithRack.size < brokers.size =>\n+        brokers.map(broker => BrokerMetadata(broker.id, None))\n+      case _ => brokers.map(broker => BrokerMetadata(broker.id, broker.rack))\n+    }\n+    brokerMetadatas.sortBy(_.id)\n+  }\n+\n+  /**\n+   * Creates or Updates the partition assignment for a given topic\n+   * @param topic\n+   * @param partitionReplicaAssignment\n+   * @param config\n+   * @param update\n+   */\n+  def createOrUpdateTopicPartitionAssignmentPathInZK(topic: String,\n+                                                     partitionReplicaAssignment: Map[Int, Seq[Int]],\n+                                                     config: Properties = new Properties,\n+                                                     update: Boolean = false) {\n+    validateCreateOrUpdateTopic(topic, partitionReplicaAssignment, config, update)\n+\n+    // Configs only matter if a topic is being created. Changing configs via AlterTopic is not supported\n+    if (!update) {\n+      // write out the config if there is any, this isn't transactional with the partition assignments\n+      zkClient.setOrCreateEntityConfigs(ConfigType.Topic, topic, config)\n+    }\n+\n+    // create the partition assignment\n+    writeTopicPartitionAssignment(topic, partitionReplicaAssignment, update)\n+  }\n+\n+  /**\n+   * Validate method to use before the topic creation or update\n+   * @param topic\n+   * @param partitionReplicaAssignment\n+   * @param config\n+   * @param update\n+   */\n+  def validateCreateOrUpdateTopic(topic: String,\n+                                  partitionReplicaAssignment: Map[Int, Seq[Int]],\n+                                  config: Properties,\n+                                  update: Boolean): Unit = {\n+    // validate arguments\n+    Topic.validate(topic)\n+\n+    if (!update) {\n+      if (zkClient.topicExists(topic))\n+        throw new TopicExistsException(s\"Topic '$topic' already exists.\")","path":"core/src/main/scala/kafka/zk/AdminZkClient.scala","position":116,"original_position":116,"commit_id":"a78fd3e0bcad8dea8707076db815a6dd589f52c7","original_commit_id":"a78fd3e0bcad8dea8707076db815a6dd589f52c7","user":{"login":"junrao","id":132330,"avatar_url":"https://avatars2.githubusercontent.com/u/132330?v=4","gravatar_id":"","url":"https://api.github.com/users/junrao","html_url":"https://github.com/junrao","followers_url":"https://api.github.com/users/junrao/followers","following_url":"https://api.github.com/users/junrao/following{/other_user}","gists_url":"https://api.github.com/users/junrao/gists{/gist_id}","starred_url":"https://api.github.com/users/junrao/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/junrao/subscriptions","organizations_url":"https://api.github.com/users/junrao/orgs","repos_url":"https://api.github.com/users/junrao/repos","events_url":"https://api.github.com/users/junrao/events{/privacy}","received_events_url":"https://api.github.com/users/junrao/received_events","type":"User","site_admin":false},"body":"Do we need this check here since writeTopicPartitionAssignment() already covers it?","created_at":"2017-11-17T19:12:02Z","updated_at":"2017-11-17T19:58:46Z","html_url":"https://github.com/apache/kafka/pull/4194#discussion_r151766517","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/4194","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/151766517"},"html":{"href":"https://github.com/apache/kafka/pull/4194#discussion_r151766517"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/4194"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/4194","id":151457918,"html_url":"https://github.com/apache/kafka/pull/4194","diff_url":"https://github.com/apache/kafka/pull/4194.diff","patch_url":"https://github.com/apache/kafka/pull/4194.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/4194","number":4194,"state":"open","locked":false,"title":"KAFKA-5646:  Use KafkaZkClient in DynamicConfigManager and AdminManager","user":{"login":"omkreddy","id":8134545,"avatar_url":"https://avatars2.githubusercontent.com/u/8134545?v=4","gravatar_id":"","url":"https://api.github.com/users/omkreddy","html_url":"https://github.com/omkreddy","followers_url":"https://api.github.com/users/omkreddy/followers","following_url":"https://api.github.com/users/omkreddy/following{/other_user}","gists_url":"https://api.github.com/users/omkreddy/gists{/gist_id}","starred_url":"https://api.github.com/users/omkreddy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/omkreddy/subscriptions","organizations_url":"https://api.github.com/users/omkreddy/orgs","repos_url":"https://api.github.com/users/omkreddy/repos","events_url":"https://api.github.com/users/omkreddy/events{/privacy}","received_events_url":"https://api.github.com/users/omkreddy/received_events","type":"User","site_admin":false},"body":"* Add AdminZkClient class\r\n* Use KafkaZkClient, AdminZkClient  in ConfigCommand, TopicCommand\r\n* All the existing tests should work \r\n\r\n### Committer Checklist\r\n- [ ] Verify design and implementation \r\n- [ ] Verify test coverage and CI build status\r\n- [ ] Verify documentation (including upgrade notes)","created_at":"2017-11-08T18:20:23Z","updated_at":"2017-11-17T19:58:46Z","closed_at":null,"merged_at":null,"merge_commit_sha":"921bea81d60e7027c7776305e0a211c9fb995a47","assignee":null,"assignees":[],"requested_reviewers":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/4194/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/4194/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/4194/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/a78fd3e0bcad8dea8707076db815a6dd589f52c7","head":{"label":"omkreddy:KAFKA-5646-ZK-ADMIN-UTILS-DYNAMIC-MANAGER","ref":"KAFKA-5646-ZK-ADMIN-UTILS-DYNAMIC-MANAGER","sha":"a78fd3e0bcad8dea8707076db815a6dd589f52c7","user":{"login":"omkreddy","id":8134545,"avatar_url":"https://avatars2.githubusercontent.com/u/8134545?v=4","gravatar_id":"","url":"https://api.github.com/users/omkreddy","html_url":"https://github.com/omkreddy","followers_url":"https://api.github.com/users/omkreddy/followers","following_url":"https://api.github.com/users/omkreddy/following{/other_user}","gists_url":"https://api.github.com/users/omkreddy/gists{/gist_id}","starred_url":"https://api.github.com/users/omkreddy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/omkreddy/subscriptions","organizations_url":"https://api.github.com/users/omkreddy/orgs","repos_url":"https://api.github.com/users/omkreddy/repos","events_url":"https://api.github.com/users/omkreddy/events{/privacy}","received_events_url":"https://api.github.com/users/omkreddy/received_events","type":"User","site_admin":false},"repo":{"id":51066257,"name":"kafka","full_name":"omkreddy/kafka","owner":{"login":"omkreddy","id":8134545,"avatar_url":"https://avatars2.githubusercontent.com/u/8134545?v=4","gravatar_id":"","url":"https://api.github.com/users/omkreddy","html_url":"https://github.com/omkreddy","followers_url":"https://api.github.com/users/omkreddy/followers","following_url":"https://api.github.com/users/omkreddy/following{/other_user}","gists_url":"https://api.github.com/users/omkreddy/gists{/gist_id}","starred_url":"https://api.github.com/users/omkreddy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/omkreddy/subscriptions","organizations_url":"https://api.github.com/users/omkreddy/orgs","repos_url":"https://api.github.com/users/omkreddy/repos","events_url":"https://api.github.com/users/omkreddy/events{/privacy}","received_events_url":"https://api.github.com/users/omkreddy/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/omkreddy/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/omkreddy/kafka","forks_url":"https://api.github.com/repos/omkreddy/kafka/forks","keys_url":"https://api.github.com/repos/omkreddy/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/omkreddy/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/omkreddy/kafka/teams","hooks_url":"https://api.github.com/repos/omkreddy/kafka/hooks","issue_events_url":"https://api.github.com/repos/omkreddy/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/omkreddy/kafka/events","assignees_url":"https://api.github.com/repos/omkreddy/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/omkreddy/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/omkreddy/kafka/tags","blobs_url":"https://api.github.com/repos/omkreddy/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/omkreddy/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/omkreddy/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/omkreddy/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/omkreddy/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/omkreddy/kafka/languages","stargazers_url":"https://api.github.com/repos/omkreddy/kafka/stargazers","contributors_url":"https://api.github.com/repos/omkreddy/kafka/contributors","subscribers_url":"https://api.github.com/repos/omkreddy/kafka/subscribers","subscription_url":"https://api.github.com/repos/omkreddy/kafka/subscription","commits_url":"https://api.github.com/repos/omkreddy/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/omkreddy/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/omkreddy/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/omkreddy/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/omkreddy/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/omkreddy/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/omkreddy/kafka/merges","archive_url":"https://api.github.com/repos/omkreddy/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/omkreddy/kafka/downloads","issues_url":"https://api.github.com/repos/omkreddy/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/omkreddy/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/omkreddy/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/omkreddy/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/omkreddy/kafka/labels{/name}","releases_url":"https://api.github.com/repos/omkreddy/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/omkreddy/kafka/deployments","created_at":"2016-02-04T09:42:20Z","updated_at":"2017-10-15T02:28:40Z","pushed_at":"2017-11-17T14:53:09Z","git_url":"git://github.com/omkreddy/kafka.git","ssh_url":"git@github.com:omkreddy/kafka.git","clone_url":"https://github.com/omkreddy/kafka.git","svn_url":"https://github.com/omkreddy/kafka","homepage":null,"size":58416,"stargazers_count":1,"watchers_count":1,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":1,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"ad71b9d068f63b2e64bcce72c557115efd2cc1ca","user":{"login":"apache","id":47359,"avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","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://avatars0.githubusercontent.com/u/47359?v=4","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}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2017-11-17T15:58:25Z","pushed_at":"2017-11-17T19:55:25Z","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":59705,"stargazers_count":6697,"watchers_count":6697,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":4061,"mirror_url":"git://git.apache.org/kafka.git","archived":false,"open_issues_count":485,"forks":4061,"open_issues":485,"watchers":6697,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/4194"},"html":{"href":"https://github.com/apache/kafka/pull/4194"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/4194"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/4194/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/4194/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/4194/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/a78fd3e0bcad8dea8707076db815a6dd589f52c7"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"junrao"}}