{"action":"created","comment":{"url":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/35438647","id":35438647,"diff_hunk":"@@ -241,3 +242,122 @@ def produce(self, messages):\n         # only *some* messages when a leader changes. Therefore, we don't want\n         # a random partition distribution changing that on the retry.\n         self._produce(self._partition_messages(messages), 0)\n+\n+\n+class AsyncProducer(Producer):\n+    \"\"\"\n+    This class implements the asynchronous producer logic found in the\n+    JVM driver. In inherits from the synchronous implementation.\n+    \"\"\"\n+    def __init__(self,\n+                 cluster,\n+                 topic,\n+                 partitioner=random_partitioner,\n+                 compression=CompressionType.NONE,\n+                 max_retries=3,\n+                 retry_backoff_ms=100,\n+                 required_acks=1,\n+                 ack_timeout_ms=10000,\n+                 batch_size=200,\n+                 max_queued_requests=500):\n+        \"\"\"Instantiate a new AsyncProducer\n+\n+        :param cluster: The cluster to which to connect\n+        :type cluster: :class:`pykafka.cluster.Cluster`\n+        :param topic: The topic to which to produce messages\n+        :type topic: :class:`pykafka.topic.Topic`\n+        :param partitioner: The partitioner to use during message production\n+        :type partitioner: :class:`pykafka.partitioners.BasePartitioner`\n+        :param compression: The type of compression to use.\n+        :type compression: :class:`pykafka.common.CompressionType`\n+        :param max_retries: How many times to attempt to produce messages\n+            before raising an error.\n+        :type max_retries: int\n+        :param retry_backoff_ms: The amount of time (in milliseconds) to\n+            back off during produce request retries.\n+        :type retry_backoff_ms: int\n+        :param required_acks: How many other brokers must have committed the\n+            data to their log and acknowledged this to the leader before a\n+            request is considered complete?\n+        :type required_acks: int\n+        :param ack_timeout_ms: Amount of time (in milliseconds) to wait for\n+            acknowledgment of a produce request.\n+        :type ack_timeout_ms: int\n+        :param batch_size: Size (in bytes) of batches to send to brokers.\n+        :type batch_size: int\n+        :param max_queued_requests: The maximum number of requests allowed\n+            in the request queue\n+        :type max_queued_requests: int\n+        \"\"\"\n+        super(AsyncProducer, self).__init__(\n+            cluster,\n+            topic,\n+            partitioner=partitioner,\n+            compression=compression,\n+            max_retries=max_retries,\n+            retry_backoff_ms=retry_backoff_ms,\n+            required_acks=required_acks,\n+            ack_timeout_ms=ack_timeout_ms,\n+            batch_size=batch_size\n+        )\n+        self._max_queued_requests = max_queued_requests\n+        self._setup_workers()\n+\n+    def _setup_workers(self):\n+        \"\"\"Spawn workers for connected brokers\n+\n+        Creates one :class:`queue.Queue` instance and one worker thread per\n+        broker. The queue holds requests waiting to be sent, and the worker\n+        thread reads the end of the queue and sends requests.\n+        \"\"\"\n+        self._queues_by_leader = {}\n+        for partition in self._topic.partitions.values():\n+            if partition.leader.id not in self._queues_by_leader:\n+                self._queues_by_leader[partition.leader.id] = self._setup_queue(partition.leader)\n+\n+    def _setup_queue(self, broker):\n+        \"\"\"Spawn a worker for the given broker\n+\n+        Creates a :class:`queue.Queue` instance and a worker thread for the\n+        given broker.\n+\n+        :param broker: The broker for which to instantiate this queue and worker\n+        :type broker: :class:`pykafka.broker.Broker`\n+        \"\"\"\n+        def worker():\n+            while True:\n+                time.sleep(.0001)\n+                try:\n+                    request, attempt = request_queue.get_nowait()\n+                    self._send_request(broker, request, attempt)\n+                except Empty:\n+                    continue\n+                except ProduceFailureError as e:\n+                    log.error(\"Producer error: %s\", e)","path":"pykafka/producer.py","position":167,"original_position":167,"commit_id":"cff64efbed3be61151fea358c86c6023fa23298e","original_commit_id":"cff64efbed3be61151fea358c86c6023fa23298e","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":"This doesn't seem right to me. Why would we log the error and then just continue on? Signaling failure from the thread is annoying, but I think this needs to be bubbled up to the user.","created_at":"2015-07-24T16:21:26Z","updated_at":"2015-07-24T16:21:26Z","html_url":"https://github.com/Parsely/pykafka/pull/177#discussion_r35438647","pull_request_url":"https://api.github.com/repos/Parsely/pykafka/pulls/177","_links":{"self":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/35438647"},"html":{"href":"https://github.com/Parsely/pykafka/pull/177#discussion_r35438647"},"pull_request":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/177"}}},"pull_request":{"url":"https://api.github.com/repos/Parsely/pykafka/pulls/177","id":36883989,"html_url":"https://github.com/Parsely/pykafka/pull/177","diff_url":"https://github.com/Parsely/pykafka/pull/177.diff","patch_url":"https://github.com/Parsely/pykafka/pull/177.patch","issue_url":"https://api.github.com/repos/Parsely/pykafka/issues/177","number":177,"state":"open","locked":false,"title":"Asynchronous Producer","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 subclasses `Producer` to implement `AsyncProducer`, which uses a thread per broker instead of the main thread for all brokers. It still assembles requests synchronously, but it puts them onto a queue and sends them asynchronously.\r\n\r\nHere are some benchmarks from a test topic on Parsely's Kafka 0.8 cluster. I ran the synchronous producer and the asynchronous producer under the same conditions, one after another. Each was running this code:\r\n\r\n```python\r\nclient = KafkaClient(hosts='parsely.kafka.hosts:9092')\r\ntopic = client.topics['emmett.test.async']\r\n\r\nproducer = topic.get_producer(\r\n    compression=CompressionType.SNAPPY\r\n)\r\n\r\nstart = time.time()\r\nnext_print = start + 5\r\ncount = 0\r\nbatch_range = 20\r\nwhile True:\r\n    producer.produce([\"msg {} {}\".format(count, i) for i in xrange(batch_range)])\r\n    count += batch_range\r\n    now = time.time()\r\n    if now > next_print:\r\n        dur = now - start\r\n        print '{} messages per second'.format(count / dur)\r\n        count = 0\r\n        start = time.time()\r\n        next_print = start + 5\r\n```\r\nThis test yielded the following results:\r\n```\r\nProducer\r\n6627.89380968 messages per second\r\n7249.33380739 messages per second\r\n6692.9103782 messages per second\r\n6543.31825874 messages per second\r\n6030.73958914 messages per second\r\n\r\nAsyncProducer\r\n18607.9476495 messages per second\r\n16925.315712 messages per second\r\n18018.9328616 messages per second\r\n19469.3402186 messages per second\r\n15855.3717282 messages per second\r\n```\r\nThis represents a 2.68x increase in average message production throughput. This is probably slightly misleading, because the number is not calculated when the request is actually made, only when it is enqueued to be picked up by the worker thread.\r\n\r\nReady for initial review from @kbourgoin @yungchin.","created_at":"2015-06-03T22:13:03Z","updated_at":"2015-07-24T16:21:26Z","closed_at":null,"merged_at":null,"merge_commit_sha":"ed3640109b69699beade2399ca0f7913b9e8ae43","assignee":{"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},"milestone":{"url":"https://api.github.com/repos/Parsely/pykafka/milestones/5","html_url":"https://github.com/Parsely/pykafka/milestones/1.2.0","labels_url":"https://api.github.com/repos/Parsely/pykafka/milestones/5/labels","id":1141520,"number":5,"title":"1.2.0","description":"Release including librdkafka, and asynchronous producer, and probably some other new stuff.","creator":{"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},"open_issues":4,"closed_issues":4,"state":"open","created_at":"2015-05-31T03:18:21Z","updated_at":"2015-07-23T23:01:14Z","due_on":null,"closed_at":null},"commits_url":"https://api.github.com/repos/Parsely/pykafka/pulls/177/commits","review_comments_url":"https://api.github.com/repos/Parsely/pykafka/pulls/177/comments","review_comment_url":"https://api.github.com/repos/Parsely/pykafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/Parsely/pykafka/issues/177/comments","statuses_url":"https://api.github.com/repos/Parsely/pykafka/statuses/cff64efbed3be61151fea358c86c6023fa23298e","head":{"label":"Parsely:feature/async-producer","ref":"feature/async-producer","sha":"cff64efbed3be61151fea358c86c6023fa23298e","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-07-24T07:52:12Z","pushed_at":"2015-07-24T16:09:07Z","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://pykafka.readthedocs.org/","size":3082,"stargazers_count":134,"watchers_count":134,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":37,"mirror_url":null,"open_issues_count":11,"forks":37,"open_issues":11,"watchers":134,"default_branch":"master"}},"base":{"label":"Parsely:master","ref":"master","sha":"1c2153c4ff444750c490f174d7f0df62ac65fae1","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-07-24T07:52:12Z","pushed_at":"2015-07-24T16:09:07Z","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://pykafka.readthedocs.org/","size":3082,"stargazers_count":134,"watchers_count":134,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":37,"mirror_url":null,"open_issues_count":11,"forks":37,"open_issues":11,"watchers":134,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/177"},"html":{"href":"https://github.com/Parsely/pykafka/pull/177"},"issue":{"href":"https://api.github.com/repos/Parsely/pykafka/issues/177"},"comments":{"href":"https://api.github.com/repos/Parsely/pykafka/issues/177/comments"},"review_comments":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/177/comments"},"review_comment":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/177/commits"},"statuses":{"href":"https://api.github.com/repos/Parsely/pykafka/statuses/cff64efbed3be61151fea358c86c6023fa23298e"}}}}