{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/77806141","id":77806141,"diff_hunk":"@@ -0,0 +1,219 @@\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.server\n+\n+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}\n+import kafka.common.TopicAndPartition\n+import kafka.server.Constants._\n+import kafka.server.ReplicationQuotaManagerConfig._\n+import kafka.utils.CoreUtils._\n+import kafka.utils.{CoreUtils, Logging}","path":"core/src/main/scala/kafka/server/ReplicationQuotaManager.scala","position":24,"original_position":24,"commit_id":"a3ab4c875b489e392870e30e6680936078f7f436","original_commit_id":"be2ca7dc178d7a102fedc9f5e990d68e64c5c26c","user":{"login":"benstopford","id":1297498,"avatar_url":"https://avatars.githubusercontent.com/u/1297498?v=3","gravatar_id":"","url":"https://api.github.com/users/benstopford","html_url":"https://github.com/benstopford","followers_url":"https://api.github.com/users/benstopford/followers","following_url":"https://api.github.com/users/benstopford/following{/other_user}","gists_url":"https://api.github.com/users/benstopford/gists{/gist_id}","starred_url":"https://api.github.com/users/benstopford/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benstopford/subscriptions","organizations_url":"https://api.github.com/users/benstopford/orgs","repos_url":"https://api.github.com/users/benstopford/repos","events_url":"https://api.github.com/users/benstopford/events{/privacy}","received_events_url":"https://api.github.com/users/benstopford/received_events","type":"User","site_admin":false},"body":"thanks","created_at":"2016-09-07T11:44:55Z","updated_at":"2016-09-07T11:44:55Z","html_url":"https://github.com/apache/kafka/pull/1776#discussion_r77806141","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/1776","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/77806141"},"html":{"href":"https://github.com/apache/kafka/pull/1776#discussion_r77806141"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/1776"}}},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/1776","id":82445483,"html_url":"https://github.com/apache/kafka/pull/1776","diff_url":"https://github.com/apache/kafka/pull/1776.diff","patch_url":"https://github.com/apache/kafka/pull/1776.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/1776","number":1776,"state":"open","locked":false,"title":"KIP-73 - Replication Quotas","user":{"login":"benstopford","id":1297498,"avatar_url":"https://avatars.githubusercontent.com/u/1297498?v=3","gravatar_id":"","url":"https://api.github.com/users/benstopford","html_url":"https://github.com/benstopford","followers_url":"https://api.github.com/users/benstopford/followers","following_url":"https://api.github.com/users/benstopford/following{/other_user}","gists_url":"https://api.github.com/users/benstopford/gists{/gist_id}","starred_url":"https://api.github.com/users/benstopford/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benstopford/subscriptions","organizations_url":"https://api.github.com/users/benstopford/orgs","repos_url":"https://api.github.com/users/benstopford/repos","events_url":"https://api.github.com/users/benstopford/events{/privacy}","received_events_url":"https://api.github.com/users/benstopford/received_events","type":"User","site_admin":false},"body":"This applies to Replication Quotas \r\nbased on KIP-73 [(link)](https://cwiki.apache.org/confluence/display/KAFKA/KIP-73+Replication+Quotas) originally motivated by KAFKA-1464.\r\n\r\n**This first PR demonstrates the approach**. \r\n\r\n**_Overview of Change_**\r\nThe guts of this change are relatively small. Throttling occurs on both leader and follower sides. A single class tracks the throttled throughput in and out of each broker (**_ReplicationQuotaManager_**).  \r\n\r\nOn the follower side, the Follower Throttled Rate is calculated as fetch responses arrive. Then, before the next fetch request is sent, we check to see if the quota is violated, removing throttled partitions from the request if it is. This is all encapsulated in a few lines of code in the **_ReplicaFetcherThread_**. There is existing code to handle temporal back off if the request ends up being empty. \r\n\r\nOn the leader side it's a little more complex. When a fetch request arrives in the leader, it is built, partition by partition, in **_ReplicaManager.readFromLocalLog_**. As we put each partition into the fetch response, we check if the total size fits in the current quota. If the quota is exceeded, the partition will not be added to the fetch response. Importantly, we don't increase the quota at this point, we just check to see if the bytes will fit. \r\n\r\nNow, if there aren't enough bytes to send the response immediately, which is common if we're catching up and throttled, then the request will be put in purgatory. I've added some simple code to **_DelayedFetch_** to handle throttled partitions (throttled partitions are checked against the quota, rather than comparing fetchMinBytes).\r\n\r\nWhen the delayed fetch completes, and exits purgatory, _**ReplicaManager.readFromLocalLog**_ will be called again. This is why _**ReplicaManager.readFromLocalLog**_ does not actually increase the quota, it just checks whether enough bytes are available for a partition. \r\n\r\nFinally, when there are enough bytes to be sent, or the delayed fetch times out, the response will be sent. Before it is sent the throttled-outbound-rate is increased, based on the size of throttled partitions being sent. This is at the end of _**KafkaApis.handleFetchRequest**_, exactly where client quotas are recorded. \r\n\r\nThere is an acceptance test which asserts the whole throttling process stabilises on the desired value. This is relatively simple at this point (two brokers only) and will need some tweaking before it can be merged, but outlines the testing done so far. See **_ReplicationQuotaTest_**.\r\n\r\nNote: \r\nIt should be noted that this protocol can over-request. The request is built, based on the quota at time t1 (_ReplicaManager.readFromLocalLog_). The bytes in the response are recorded at time t2 (end of _KafkaApis.handleFetchRequest_), where t2 > t1. For this reason I originally included an OverRequestedRate as a JMX metric, but in testing I'm not seeing this cause an issue. Over-requesting is quickly compensated by subsequent requests, stabilising close to the quota value, but I may alter the approach used here as I do more testing. \r\n\r\n_**Main stuff left to do:**_\r\n- The fetch size is currently unbounded. This will be addressed in KIP-74, but we need to ensure this ensures requests don’t go beyond the throttle window.\r\n- There is no specific validation or other utilities in ConfigCommand or code in ReassignPartitionsCommand for validation etc.\r\n- Tests are largely happy path, the functional tests rely on a single broker-to-broker set up. There are no system tests. ","created_at":"2016-08-23T21:35:20Z","updated_at":"2016-09-07T11:44:55Z","closed_at":null,"merged_at":null,"merge_commit_sha":"86c8f43afee1b94e423e6eed09b652e715b2cca6","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/1776/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/1776/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/1776/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/a3ab4c875b489e392870e30e6680936078f7f436","head":{"label":"benstopford:rep-quotas-v2","ref":"rep-quotas-v2","sha":"a3ab4c875b489e392870e30e6680936078f7f436","user":{"login":"benstopford","id":1297498,"avatar_url":"https://avatars.githubusercontent.com/u/1297498?v=3","gravatar_id":"","url":"https://api.github.com/users/benstopford","html_url":"https://github.com/benstopford","followers_url":"https://api.github.com/users/benstopford/followers","following_url":"https://api.github.com/users/benstopford/following{/other_user}","gists_url":"https://api.github.com/users/benstopford/gists{/gist_id}","starred_url":"https://api.github.com/users/benstopford/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benstopford/subscriptions","organizations_url":"https://api.github.com/users/benstopford/orgs","repos_url":"https://api.github.com/users/benstopford/repos","events_url":"https://api.github.com/users/benstopford/events{/privacy}","received_events_url":"https://api.github.com/users/benstopford/received_events","type":"User","site_admin":false},"repo":{"id":40359697,"name":"kafka","full_name":"benstopford/kafka","owner":{"login":"benstopford","id":1297498,"avatar_url":"https://avatars.githubusercontent.com/u/1297498?v=3","gravatar_id":"","url":"https://api.github.com/users/benstopford","html_url":"https://github.com/benstopford","followers_url":"https://api.github.com/users/benstopford/followers","following_url":"https://api.github.com/users/benstopford/following{/other_user}","gists_url":"https://api.github.com/users/benstopford/gists{/gist_id}","starred_url":"https://api.github.com/users/benstopford/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/benstopford/subscriptions","organizations_url":"https://api.github.com/users/benstopford/orgs","repos_url":"https://api.github.com/users/benstopford/repos","events_url":"https://api.github.com/users/benstopford/events{/privacy}","received_events_url":"https://api.github.com/users/benstopford/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/benstopford/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/benstopford/kafka","forks_url":"https://api.github.com/repos/benstopford/kafka/forks","keys_url":"https://api.github.com/repos/benstopford/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/benstopford/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/benstopford/kafka/teams","hooks_url":"https://api.github.com/repos/benstopford/kafka/hooks","issue_events_url":"https://api.github.com/repos/benstopford/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/benstopford/kafka/events","assignees_url":"https://api.github.com/repos/benstopford/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/benstopford/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/benstopford/kafka/tags","blobs_url":"https://api.github.com/repos/benstopford/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/benstopford/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/benstopford/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/benstopford/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/benstopford/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/benstopford/kafka/languages","stargazers_url":"https://api.github.com/repos/benstopford/kafka/stargazers","contributors_url":"https://api.github.com/repos/benstopford/kafka/contributors","subscribers_url":"https://api.github.com/repos/benstopford/kafka/subscribers","subscription_url":"https://api.github.com/repos/benstopford/kafka/subscription","commits_url":"https://api.github.com/repos/benstopford/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/benstopford/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/benstopford/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/benstopford/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/benstopford/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/benstopford/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/benstopford/kafka/merges","archive_url":"https://api.github.com/repos/benstopford/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/benstopford/kafka/downloads","issues_url":"https://api.github.com/repos/benstopford/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/benstopford/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/benstopford/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/benstopford/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/benstopford/kafka/labels{/name}","releases_url":"https://api.github.com/repos/benstopford/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/benstopford/kafka/deployments","created_at":"2015-08-07T12:36:48Z","updated_at":"2016-01-25T13:54:01Z","pushed_at":"2016-09-07T11:40:34Z","git_url":"git://github.com/benstopford/kafka.git","ssh_url":"git@github.com:benstopford/kafka.git","clone_url":"https://github.com/benstopford/kafka.git","svn_url":"https://github.com/benstopford/kafka","homepage":null,"size":32231,"stargazers_count":0,"watchers_count":0,"language":"Java","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":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"b4ddd021b42225f8cb1d7049298f0e503cb09633","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}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2016-09-07T10:00:43Z","pushed_at":"2016-09-07T11:40:35Z","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":47169,"stargazers_count":3719,"watchers_count":3719,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2310,"mirror_url":"git://git.apache.org/kafka.git","open_issues_count":247,"forks":2310,"open_issues":247,"watchers":3719,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/1776"},"html":{"href":"https://github.com/apache/kafka/pull/1776"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/1776"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/1776/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/1776/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/1776/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/a3ab4c875b489e392870e30e6680936078f7f436"}}}}
{"actor":{"display_login":"benstopford"}}