{"action":"created","comment":{"url":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/28159055","id":28159055,"diff_hunk":"@@ -0,0 +1,371 @@\n+import itertools\n+import logging as log\n+import signal\n+import socket\n+import sys\n+import time\n+from uuid import uuid4\n+import weakref\n+\n+from kazoo.client import KazooClient\n+from kazoo.exceptions import NoNodeException, NodeExistsError\n+from kazoo.recipe.watchers import ChildrenWatch\n+\n+from kafka.common import OffsetType\n+from kafka.pykafka.simpleconsumer import SimpleConsumer\n+\n+\n+class BalancedConsumer():\n+    def __init__(self,\n+                 topic,\n+                 cluster,\n+                 consumer_group,\n+                 fetch_message_max_bytes=1024 * 1024,\n+                 num_consumer_fetchers=1,\n+                 auto_commit_enable=False,\n+                 auto_commit_interval_ms=60 * 1000,\n+                 queued_max_messages=2000,\n+                 fetch_min_bytes=1,\n+                 fetch_wait_max_ms=100,\n+                 refresh_leader_backoff_ms=200,\n+                 offsets_channel_backoff_ms=1000,\n+                 offsets_commit_max_retries=5,\n+                 auto_offset_reset=OffsetType.LATEST,\n+                 consumer_timeout_ms=-1,\n+                 rebalance_max_retries=5,\n+                 rebalance_backoff_ms=2 * 1000,\n+                 zookeeper_connection_timeout_ms=6 * 1000,\n+                 zookeeper_connect='127.0.0.1:2181'):\n+        \"\"\"Create a BalancedConsumer\n+\n+        Maintains a single instance of SimpleConsumer, periodically using the\n+        consumer rebalancing algorithm to reassign partitions to this\n+        SimpleConsumer.\n+\n+        :param topic: the topic this consumer should consume\n+        :type topic: pykafka.topic.Topic\n+        :param cluster: the cluster this consumer should connect to\n+        :type cluster: pykafka.cluster.Cluster\n+        :param consumer_group: the name of the consumer group to join\n+        :type consumer_group: str\n+        :param fetch_message_max_bytes: the number of bytes of messages to\n+            attempt to fetch\n+        :type fetch_message_max_bytes: int\n+        :param num_consumer_fetchers: the number of threads used to fetch data\n+        :type num_consumer_fetchers: int\n+        :param auto_commit_enable: if true, periodically commit to kafka the\n+            offset of messages already fetched by this consumer\n+        :type auto_commit_enable: bool\n+        :param auto_commit_interval_ms: the frequency in ms that the consumer\n+            offsets are committed to kafka\n+        :type auto_commit_interval_ms: int\n+        :param queued_max_messages: max number of messages buffered for\n+            consumption\n+        :type queued_max_messages: int\n+        :param fetch_min_bytes: the minimum amount of data the server should\n+            return for a fetch request. If insufficient data is available the\n+            request will block\n+        :type fetch_min_bytes: int\n+        :param fetch_wait_max_ms: the maximum amount of time the server will\n+            block before answering the fetch request if there isn't sufficient\n+            data to immediately satisfy fetch_min_bytes\n+        :type fetch_wait_max_ms: int\n+        :param refresh_leader_backoff_ms: backoff time to refresh the leader of\n+            a partition after it loses the current leader\n+        :type refresh_leader_backoff_ms: int\n+        :param offsets_channel_backoff_ms: backoff time to retry offset\n+            commits/fetches\n+        :type offsets_channel_backoff_ms: int\n+        :param offsets_commit_max_retries: Retry the offset commit up to this\n+            many times on failure.\n+        :type offsets_commit_max_retries: int\n+        :param auto_offset_reset: what to do if an offset is out of range\n+        :type auto_offset_reset: int\n+        :param consumer_timeout_ms: throw a timeout exception to the consumer\n+            if no message is available for consumption after the specified\n+            interval\n+        :type consumer_timeout_ms: int\n+        :param rebalance_max_retries: Maximum number of attempts before failing to\n+            rebalance\n+        :type rebalance_max_retries: int\n+        :param rebalance_backoff_ms: Backoff time between retries during rebalance.\n+        :type rebalance_backoff_ms: int\n+        :param zookeeper_connection_timeout_ms: The max time that the client\n+            waits while establishing a connection to zookeeper.\n+        :type zookeeper_connection_timeout_ms: int\n+        :param zookeeper_connect: comma separated ip:port strings of the\n+            zookeeper nodes to connect to\n+        :type zookeeper_connect: str\n+        \"\"\"\n+        if not isinstance(cluster, weakref.ProxyType):\n+            self._cluster = weakref.proxy(cluster)\n+        else:\n+            self._cluster = cluster\n+        self._consumer_group = consumer_group\n+        self._topic = topic\n+\n+        self._auto_commit_enable = auto_commit_enable\n+        self._auto_commit_interval_ms = auto_commit_interval_ms\n+        self._fetch_message_max_bytes = fetch_message_max_bytes\n+        self._fetch_min_bytes = fetch_min_bytes\n+        self._rebalance_max_retries = rebalance_max_retries\n+        self._num_consumer_fetchers = num_consumer_fetchers\n+        self._queued_max_messages = queued_max_messages\n+        self._fetch_wait_max_ms = fetch_wait_max_ms\n+        self._rebalance_backoff_ms = rebalance_backoff_ms\n+        self._consumer_timeout_ms = consumer_timeout_ms\n+\n+        self._consumer = None\n+        self._consumer_id = \"{}:{}\".format(socket.gethostname(), uuid4())\n+        self._partitions = set()\n+        self._setting_watches = True\n+\n+        self._topic_path = '/consumers/{}/owners/{}'.format(self._consumer_group,\n+                                                            self._topic.name)\n+        self._consumer_id_path = '/consumers/{}/ids'.format(self._consumer_group)\n+\n+        def _close_zk_connection(signum, frame):\n+            self._zookeeper.stop()\n+            self._consumer.stop()\n+            sys.exit()\n+        signal.signal(signal.SIGINT, _close_zk_connection)","path":"kafka/pykafka/balancedconsumer.py","position":131,"original_position":131,"commit_id":"5a516c352172914d25786b594af7929af087ca8c","original_commit_id":"5a516c352172914d25786b594af7929af087ca8c","user":{"login":"kbourgoin","id":136574,"avatar_url":"https://avatars.githubusercontent.com/u/136574?v=3","gravatar_id":"","url":"https://api.github.com/users/kbourgoin","html_url":"https://github.com/kbourgoin","followers_url":"https://api.github.com/users/kbourgoin/followers","following_url":"https://api.github.com/users/kbourgoin/following{/other_user}","gists_url":"https://api.github.com/users/kbourgoin/gists{/gist_id}","starred_url":"https://api.github.com/users/kbourgoin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kbourgoin/subscriptions","organizations_url":"https://api.github.com/users/kbourgoin/orgs","repos_url":"https://api.github.com/users/kbourgoin/repos","events_url":"https://api.github.com/users/kbourgoin/events{/privacy}","received_events_url":"https://api.github.com/users/kbourgoin/received_events","type":"User","site_admin":false},"body":"I don't think we want to catch SIGINT. This is going to throw an exception if someone tries to run a consumer in the background thread.","created_at":"2015-04-10T16:26:12Z","updated_at":"2015-04-10T16:26:12Z","html_url":"https://github.com/Parsely/pykafka/pull/138#discussion_r28159055","pull_request_url":"https://api.github.com/repos/Parsely/pykafka/pulls/138","_links":{"self":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/28159055"},"html":{"href":"https://github.com/Parsely/pykafka/pull/138#discussion_r28159055"},"pull_request":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/138"}}},"pull_request":{"url":"https://api.github.com/repos/Parsely/pykafka/pulls/138","id":32880881,"html_url":"https://github.com/Parsely/pykafka/pull/138","diff_url":"https://github.com/Parsely/pykafka/pull/138.diff","patch_url":"https://github.com/Parsely/pykafka/pull/138.patch","issue_url":"https://api.github.com/repos/Parsely/pykafka/issues/138","number":138,"state":"open","locked":false,"title":"[WIP] Balanced Consumer","user":{"login":"emmett9001","id":723615,"avatar_url":"https://avatars.githubusercontent.com/u/723615?v=3","gravatar_id":"","url":"https://api.github.com/users/emmett9001","html_url":"https://github.com/emmett9001","followers_url":"https://api.github.com/users/emmett9001/followers","following_url":"https://api.github.com/users/emmett9001/following{/other_user}","gists_url":"https://api.github.com/users/emmett9001/gists{/gist_id}","starred_url":"https://api.github.com/users/emmett9001/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/emmett9001/subscriptions","organizations_url":"https://api.github.com/users/emmett9001/orgs","repos_url":"https://api.github.com/users/emmett9001/repos","events_url":"https://api.github.com/users/emmett9001/events{/privacy}","received_events_url":"https://api.github.com/users/emmett9001/received_events","type":"User","site_admin":false},"body":"This pull request adds a `BalancedConsumer` to pykafka. Related to [this issue](https://github.com/Parsely/pykafka/issues/116).\r\n\r\nSince the minute details of actual message consumption are handled in `SimpleConsumer`, and since the `BalancedConsumer` maintains its own instance of one, the `BalancedConsumer` class essentially amounts to a 250+ line implementation of the consumer rebalancing algorithm detailed [here](http://kafka.apache.org/documentation.html).\r\n\r\nCurrently, this is the only piece of pykafka that uses a `KazooClient`.\r\n\r\nMajor changes:\r\n* Implements `BalancedConsumer`\r\n* Adds some configuration parameters to `SimpleConsumer`\r\n* Makes messages fetching thread-safe in the `SimpleConsumer`\r\n* Uses a single buffer allocation for kafka api responses\r\n* Adds convenience methods to `Topic` for creating consumers\r\n* Adds a few configuration parameters to `Cluster`\r\n\r\nTodo:\r\n* [ ] To properly support the `offsets.channel` config options, `Broker` should maintain two connections: one for non-offset-related requests and one for offset-related requests. This will allow the socket timeout to be configurable for commit requests independently of the timeout for other requests.\r\n* [ ] Support other remaining configuration options: `refresh_leader_backoff_ms`, `auto_offset_reset`, `zookeeper.session.timeout.ms`, `zookeeper.sync.time.ms`, `offsets.channel.socket.timeout.ms`, `offsets.channel.backoff.ms`, `offsets.commit.max.retries`\r\n","created_at":"2015-04-08T17:04:03Z","updated_at":"2015-04-10T16:26:12Z","closed_at":null,"merged_at":null,"merge_commit_sha":"5b808fe87390f7d9c20794321fe7feb1d72421a5","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/Parsely/pykafka/pulls/138/commits","review_comments_url":"https://api.github.com/repos/Parsely/pykafka/pulls/138/comments","review_comment_url":"https://api.github.com/repos/Parsely/pykafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/Parsely/pykafka/issues/138/comments","statuses_url":"https://api.github.com/repos/Parsely/pykafka/statuses/5a516c352172914d25786b594af7929af087ca8c","head":{"label":"Parsely:feature/balanced_consumer","ref":"feature/balanced_consumer","sha":"5a516c352172914d25786b594af7929af087ca8c","user":{"login":"Parsely","id":779641,"avatar_url":"https://avatars.githubusercontent.com/u/779641?v=3","gravatar_id":"","url":"https://api.github.com/users/Parsely","html_url":"https://github.com/Parsely","followers_url":"https://api.github.com/users/Parsely/followers","following_url":"https://api.github.com/users/Parsely/following{/other_user}","gists_url":"https://api.github.com/users/Parsely/gists{/gist_id}","starred_url":"https://api.github.com/users/Parsely/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Parsely/subscriptions","organizations_url":"https://api.github.com/users/Parsely/orgs","repos_url":"https://api.github.com/users/Parsely/repos","events_url":"https://api.github.com/users/Parsely/events{/privacy}","received_events_url":"https://api.github.com/users/Parsely/received_events","type":"Organization","site_admin":false},"repo":{"id":4978045,"name":"pykafka","full_name":"Parsely/pykafka","owner":{"login":"Parsely","id":779641,"avatar_url":"https://avatars.githubusercontent.com/u/779641?v=3","gravatar_id":"","url":"https://api.github.com/users/Parsely","html_url":"https://github.com/Parsely","followers_url":"https://api.github.com/users/Parsely/followers","following_url":"https://api.github.com/users/Parsely/following{/other_user}","gists_url":"https://api.github.com/users/Parsely/gists{/gist_id}","starred_url":"https://api.github.com/users/Parsely/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Parsely/subscriptions","organizations_url":"https://api.github.com/users/Parsely/orgs","repos_url":"https://api.github.com/users/Parsely/repos","events_url":"https://api.github.com/users/Parsely/events{/privacy}","received_events_url":"https://api.github.com/users/Parsely/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/Parsely/pykafka","description":"kafka client for python","fork":false,"url":"https://api.github.com/repos/Parsely/pykafka","forks_url":"https://api.github.com/repos/Parsely/pykafka/forks","keys_url":"https://api.github.com/repos/Parsely/pykafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/Parsely/pykafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/Parsely/pykafka/teams","hooks_url":"https://api.github.com/repos/Parsely/pykafka/hooks","issue_events_url":"https://api.github.com/repos/Parsely/pykafka/issues/events{/number}","events_url":"https://api.github.com/repos/Parsely/pykafka/events","assignees_url":"https://api.github.com/repos/Parsely/pykafka/assignees{/user}","branches_url":"https://api.github.com/repos/Parsely/pykafka/branches{/branch}","tags_url":"https://api.github.com/repos/Parsely/pykafka/tags","blobs_url":"https://api.github.com/repos/Parsely/pykafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/Parsely/pykafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/Parsely/pykafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/Parsely/pykafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/Parsely/pykafka/statuses/{sha}","languages_url":"https://api.github.com/repos/Parsely/pykafka/languages","stargazers_url":"https://api.github.com/repos/Parsely/pykafka/stargazers","contributors_url":"https://api.github.com/repos/Parsely/pykafka/contributors","subscribers_url":"https://api.github.com/repos/Parsely/pykafka/subscribers","subscription_url":"https://api.github.com/repos/Parsely/pykafka/subscription","commits_url":"https://api.github.com/repos/Parsely/pykafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/Parsely/pykafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/Parsely/pykafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/Parsely/pykafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/Parsely/pykafka/contents/{+path}","compare_url":"https://api.github.com/repos/Parsely/pykafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/Parsely/pykafka/merges","archive_url":"https://api.github.com/repos/Parsely/pykafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/Parsely/pykafka/downloads","issues_url":"https://api.github.com/repos/Parsely/pykafka/issues{/number}","pulls_url":"https://api.github.com/repos/Parsely/pykafka/pulls{/number}","milestones_url":"https://api.github.com/repos/Parsely/pykafka/milestones{/number}","notifications_url":"https://api.github.com/repos/Parsely/pykafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/Parsely/pykafka/labels{/name}","releases_url":"https://api.github.com/repos/Parsely/pykafka/releases{/id}","created_at":"2012-07-10T18:02:21Z","updated_at":"2015-04-08T16:44:27Z","pushed_at":"2015-04-09T23:38:43Z","git_url":"git://github.com/Parsely/pykafka.git","ssh_url":"git@github.com:Parsely/pykafka.git","clone_url":"https://github.com/Parsely/pykafka.git","svn_url":"https://github.com/Parsely/pykafka","homepage":"http://samsa.readthedocs.org/","size":2196,"stargazers_count":59,"watchers_count":59,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":21,"mirror_url":null,"open_issues_count":21,"forks":21,"open_issues":21,"watchers":59,"default_branch":"master"}},"base":{"label":"Parsely:master","ref":"master","sha":"2260b33b174eb541e5517dad8bfb24fc36aec41c","user":{"login":"Parsely","id":779641,"avatar_url":"https://avatars.githubusercontent.com/u/779641?v=3","gravatar_id":"","url":"https://api.github.com/users/Parsely","html_url":"https://github.com/Parsely","followers_url":"https://api.github.com/users/Parsely/followers","following_url":"https://api.github.com/users/Parsely/following{/other_user}","gists_url":"https://api.github.com/users/Parsely/gists{/gist_id}","starred_url":"https://api.github.com/users/Parsely/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Parsely/subscriptions","organizations_url":"https://api.github.com/users/Parsely/orgs","repos_url":"https://api.github.com/users/Parsely/repos","events_url":"https://api.github.com/users/Parsely/events{/privacy}","received_events_url":"https://api.github.com/users/Parsely/received_events","type":"Organization","site_admin":false},"repo":{"id":4978045,"name":"pykafka","full_name":"Parsely/pykafka","owner":{"login":"Parsely","id":779641,"avatar_url":"https://avatars.githubusercontent.com/u/779641?v=3","gravatar_id":"","url":"https://api.github.com/users/Parsely","html_url":"https://github.com/Parsely","followers_url":"https://api.github.com/users/Parsely/followers","following_url":"https://api.github.com/users/Parsely/following{/other_user}","gists_url":"https://api.github.com/users/Parsely/gists{/gist_id}","starred_url":"https://api.github.com/users/Parsely/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/Parsely/subscriptions","organizations_url":"https://api.github.com/users/Parsely/orgs","repos_url":"https://api.github.com/users/Parsely/repos","events_url":"https://api.github.com/users/Parsely/events{/privacy}","received_events_url":"https://api.github.com/users/Parsely/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/Parsely/pykafka","description":"kafka client for python","fork":false,"url":"https://api.github.com/repos/Parsely/pykafka","forks_url":"https://api.github.com/repos/Parsely/pykafka/forks","keys_url":"https://api.github.com/repos/Parsely/pykafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/Parsely/pykafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/Parsely/pykafka/teams","hooks_url":"https://api.github.com/repos/Parsely/pykafka/hooks","issue_events_url":"https://api.github.com/repos/Parsely/pykafka/issues/events{/number}","events_url":"https://api.github.com/repos/Parsely/pykafka/events","assignees_url":"https://api.github.com/repos/Parsely/pykafka/assignees{/user}","branches_url":"https://api.github.com/repos/Parsely/pykafka/branches{/branch}","tags_url":"https://api.github.com/repos/Parsely/pykafka/tags","blobs_url":"https://api.github.com/repos/Parsely/pykafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/Parsely/pykafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/Parsely/pykafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/Parsely/pykafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/Parsely/pykafka/statuses/{sha}","languages_url":"https://api.github.com/repos/Parsely/pykafka/languages","stargazers_url":"https://api.github.com/repos/Parsely/pykafka/stargazers","contributors_url":"https://api.github.com/repos/Parsely/pykafka/contributors","subscribers_url":"https://api.github.com/repos/Parsely/pykafka/subscribers","subscription_url":"https://api.github.com/repos/Parsely/pykafka/subscription","commits_url":"https://api.github.com/repos/Parsely/pykafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/Parsely/pykafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/Parsely/pykafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/Parsely/pykafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/Parsely/pykafka/contents/{+path}","compare_url":"https://api.github.com/repos/Parsely/pykafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/Parsely/pykafka/merges","archive_url":"https://api.github.com/repos/Parsely/pykafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/Parsely/pykafka/downloads","issues_url":"https://api.github.com/repos/Parsely/pykafka/issues{/number}","pulls_url":"https://api.github.com/repos/Parsely/pykafka/pulls{/number}","milestones_url":"https://api.github.com/repos/Parsely/pykafka/milestones{/number}","notifications_url":"https://api.github.com/repos/Parsely/pykafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/Parsely/pykafka/labels{/name}","releases_url":"https://api.github.com/repos/Parsely/pykafka/releases{/id}","created_at":"2012-07-10T18:02:21Z","updated_at":"2015-04-08T16:44:27Z","pushed_at":"2015-04-09T23:38:43Z","git_url":"git://github.com/Parsely/pykafka.git","ssh_url":"git@github.com:Parsely/pykafka.git","clone_url":"https://github.com/Parsely/pykafka.git","svn_url":"https://github.com/Parsely/pykafka","homepage":"http://samsa.readthedocs.org/","size":2196,"stargazers_count":59,"watchers_count":59,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":21,"mirror_url":null,"open_issues_count":21,"forks":21,"open_issues":21,"watchers":59,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/138"},"html":{"href":"https://github.com/Parsely/pykafka/pull/138"},"issue":{"href":"https://api.github.com/repos/Parsely/pykafka/issues/138"},"comments":{"href":"https://api.github.com/repos/Parsely/pykafka/issues/138/comments"},"review_comments":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/138/comments"},"review_comment":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/138/commits"},"statuses":{"href":"https://api.github.com/repos/Parsely/pykafka/statuses/5a516c352172914d25786b594af7929af087ca8c"}}}}