{"action":"created","comment":{"url":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/37545648","id":37545648,"diff_hunk":"@@ -180,56 +331,128 @@ def _get_partition_msgs(partition_id, req):\n             attempt += 1\n             if attempt < self._max_retries:\n                 time.sleep(self._retry_backoff_ms / 1000)\n-                self._produce(to_retry, attempt)\n+                # we have to call _produce here since the broker/partition\n+                # target for each message may have changed\n+                for _ in self._produce(to_retry, attempt):\n+                    pass\n             else:\n                 raise ProduceFailureError('Unable to produce messages. See log for details.')\n \n-    def _partition_messages(self, messages):\n-        \"\"\"Assign messages to partitions using the partitioner.\n+    @raise_worker_exceptions\n+    def _update_leaders(self):\n+        \"\"\"Ensure each message in each queue is in the queue owned by its\n+            partition's leader\n \n-        :param messages: Iterable of messages to publish.\n-        :returns:        Generator of ((key, value), partition_id)\n+        This function empties all broker queues, maps their messages to the\n+        current leaders for their partitions, and enqueues the messages in\n+        the appropriate queues.\n         \"\"\"\n-        partitions = self._topic.partitions.values()\n-        for message in messages:\n-            if isinstance(message, basestring):\n-                key = None\n-                value = message\n-            else:\n-                key, value = message\n-            value = str(value)\n-            yield (key, value), self._partitioner(partitions, message).id\n+        # empty queues and figure out updated partition leaders\n+        new_queue_contents = defaultdict(list)\n+        for owned_broker in self._owned_brokers.itervalues():\n+            owned_broker.lock.acquire()\n+            current_queue_contents = owned_broker.flush(0, acquire=False,\n+                                                        release_inflight=True)\n+            for kv, partition_id in current_queue_contents:\n+                partition_leader = self._topic.partitions[partition_id].leader\n+                new_queue_contents[partition_leader.id].append((kv, partition_id))\n+        # retain locks for all brokers between these two steps\n+        for owned_broker in self._owned_brokers.itervalues():\n+            owned_broker.enqueue(new_queue_contents[owned_broker.broker.id],\n+                                 self._block_on_queue_full,\n+                                 acquire=False)\n+            owned_broker.resolve_event_state()\n+            owned_broker.lock.release()\n \n-    def _produce(self, message_partition_tups, attempt):\n-        \"\"\"Publish a set of messages to relevant brokers.\n+    @raise_worker_exceptions\n+    def _wait_all(self):\n+        \"\"\"Block until all messages in flight are sent","path":"pykafka/producer.py","position":385,"original_position":385,"commit_id":"d65f608a3bc5d0d828a12b428ac29312485b180e","original_commit_id":"d65f608a3bc5d0d828a12b428ac29312485b180e","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":"\"pending\" is much better, I don't think it's close.","created_at":"2015-08-20T15:47:32Z","updated_at":"2015-08-20T15:47:32Z","html_url":"https://github.com/Parsely/pykafka/pull/177#discussion_r37545648","pull_request_url":"https://api.github.com/repos/Parsely/pykafka/pulls/177","_links":{"self":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/37545648"},"html":{"href":"https://github.com/Parsely/pykafka/pull/177#discussion_r37545648"},"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 implements `AsyncProducer`, which uses a thread per broker instead of the main thread for all brokers.\r\n\r\nCode that can be used to test this producer's throughput:\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    for res in producer.produce([\"msg {} {}\".format(count, i) for i in xrange(batch_range)]):\r\n        pass\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\n\r\nThroughput test results with the above script (how many times per second each producer can return from `produce()`):\r\nSynchronous producer:\r\n``` \r\n66.2043259773 messages per second\r\n66.260400353 messages per second\r\n66.244598007 messages per second\r\n66.3166715681 messages per second\r\n```\r\nAsync producer:\r\n```\r\n17960.6287112 messages per second\r\n16711.675672 messages per second\r\n18904.4348297 messages per second\r\n18581.195363 messages per second\r\n19335.5599431 messages per second\r\n17676.1641205 messages per second\r\n22010.9735254 messages per second\r\n20010.7615108 messages per second\r\n```\r\n\r\nThe old synchronous producer was doing about 4500 messages per second.\r\n\r\nConnects to #124.\r\n\r\nTodo:\r\n*  [x] Monitor thread death\r\n*  [x] Abstract thread management logic into the `QueueInfo` class\r\n*  [x] Make `Producer` as small as possible by having it be a special case of `AsyncProducer`\r\n*  [x] Tests","created_at":"2015-06-03T22:13:03Z","updated_at":"2015-08-20T15:47:32Z","closed_at":null,"merged_at":null,"merge_commit_sha":"5efb9852e9daeca93c7c610a80cce754111f7d78","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/d65f608a3bc5d0d828a12b428ac29312485b180e","head":{"label":"Parsely:feature/async-producer","ref":"feature/async-producer","sha":"d65f608a3bc5d0d828a12b428ac29312485b180e","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-08-19T21:45:03Z","pushed_at":"2015-08-20T00:15:39Z","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":3329,"stargazers_count":155,"watchers_count":155,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":43,"mirror_url":null,"open_issues_count":18,"forks":43,"open_issues":18,"watchers":155,"default_branch":"master"}},"base":{"label":"Parsely:master","ref":"master","sha":"dd6997e9ee2ebe56e7c612de757fccd2a4c2f367","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-08-19T21:45:03Z","pushed_at":"2015-08-20T00:15:39Z","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":3329,"stargazers_count":155,"watchers_count":155,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":43,"mirror_url":null,"open_issues_count":18,"forks":43,"open_issues":18,"watchers":155,"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/d65f608a3bc5d0d828a12b428ac29312485b180e"}}}}