{"action":"created","comment":{"url":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/56743790","id":56743790,"diff_hunk":"@@ -540,7 +575,40 @@ def flush(self, linger_ms, release_pending=False):\n         \"\"\"\n         self._wait_for_flush_ready(linger_ms)\n         with self.lock:\n-            batch = [self.queue.pop() for _ in range(len(self.queue))]\n+            batch = []\n+            batch_size_in_bytes = 0\n+            while len(self.queue) > 0:\n+                # TODO: Should we optimistically pop and renenqueue vs. peak\n+                peaked_message = self.queue[-1]\n+\n+                if peaked_message and peaked_message.value is not None:\n+                    if len(peaked_message.value) > max_request_size:\n+                        exc = MessageSizeTooLarge(\n+                            \"Message size larger then max_request_size: {}\".format(max_request_size)\n+                        )\n+                        log.warning(exc)\n+                        # bind the MessageSizeTooLarge error the delivery\n+                        # report and remove it from the producer queue\n+                        message = self.queue.pop()\n+                        peaked_message.delivery_report_q.put((message, exc))\n+                        # remove from pending message count\n+                        self.increment_messages_pending(-1)\n+                        break\n+\n+                    # test if adding the message would go over the\n+                    # max_request_size. if it would, break out of loop\n+                    elif batch_size_in_bytes + len(peaked_message.value) > max_request_size:\n+                        log.debug(\"max_request_size reached. producing batch\")\n+                        # if we did not fully empty the queue. reset the\n+                        # flush_ready so we send another batch immediately\n+                        self.flush_ready.set()\n+                        break\n+\n+                message = self.queue.pop()\n+                if message.value is not None:\n+                    batch_size_in_bytes += len(message.value)","path":"pykafka/producer.py","position":172,"original_position":172,"commit_id":"9c81855bb99ccf24437cd6dcdac90e729d63252f","original_commit_id":"9c81855bb99ccf24437cd6dcdac90e729d63252f","user":{"login":"yungchin","id":780001,"avatar_url":"https://avatars.githubusercontent.com/u/780001?v=3","gravatar_id":"","url":"https://api.github.com/users/yungchin","html_url":"https://github.com/yungchin","followers_url":"https://api.github.com/users/yungchin/followers","following_url":"https://api.github.com/users/yungchin/following{/other_user}","gists_url":"https://api.github.com/users/yungchin/gists{/gist_id}","starred_url":"https://api.github.com/users/yungchin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/yungchin/subscriptions","organizations_url":"https://api.github.com/users/yungchin/orgs","repos_url":"https://api.github.com/users/yungchin/repos","events_url":"https://api.github.com/users/yungchin/events{/privacy}","received_events_url":"https://api.github.com/users/yungchin/received_events","type":"User","site_admin":false},"body":"Same as above here: you want `len(message)`","created_at":"2016-03-19T04:13:22Z","updated_at":"2016-03-19T04:13:22Z","html_url":"https://github.com/Parsely/pykafka/pull/480#discussion_r56743790","pull_request_url":"https://api.github.com/repos/Parsely/pykafka/pulls/480","_links":{"self":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/comments/56743790"},"html":{"href":"https://github.com/Parsely/pykafka/pull/480#discussion_r56743790"},"pull_request":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/480"}}},"pull_request":{"url":"https://api.github.com/repos/Parsely/pykafka/pulls/480","id":61755982,"html_url":"https://github.com/Parsely/pykafka/pull/480","diff_url":"https://github.com/Parsely/pykafka/pull/480.diff","patch_url":"https://github.com/Parsely/pykafka/pull/480.patch","issue_url":"https://api.github.com/repos/Parsely/pykafka/issues/480","number":480,"state":"open","locked":false,"title":"compressed message batch sizing","user":{"login":"jofusa","id":800669,"avatar_url":"https://avatars.githubusercontent.com/u/800669?v=3","gravatar_id":"","url":"https://api.github.com/users/jofusa","html_url":"https://github.com/jofusa","followers_url":"https://api.github.com/users/jofusa/followers","following_url":"https://api.github.com/users/jofusa/following{/other_user}","gists_url":"https://api.github.com/users/jofusa/gists{/gist_id}","starred_url":"https://api.github.com/users/jofusa/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jofusa/subscriptions","organizations_url":"https://api.github.com/users/jofusa/orgs","repos_url":"https://api.github.com/users/jofusa/repos","events_url":"https://api.github.com/users/jofusa/events{/privacy}","received_events_url":"https://api.github.com/users/jofusa/received_events","type":"User","site_admin":false},"body":"Just wanted some code to start a discussion.\r\n\r\nI have been struggling to deal with some `kafka.common.MessageSizeTooLargeException` I have found in my kafka logs. My messages should be well below the max message size. I am most interested in surfacing these errors in my client. My first attempt to build a producer exception background thread failed because the delivery_report queue is thread local. Also note the sync producer is too slow to process the source topic in any reasonable amount of time.\r\n\r\nSo my first question is why is the delivery_report queue is thread local. I am sure there is a good reason, but just wondering.\r\n\r\nSecond, what are thoughts on the the approach in this pr. I think if would make sense to only trigger the callback on a the list of `non_recoverable` exceptions. Really i am using this code to try to report the failed message to help diagnose the underlying cause.\r\n\r\n```py\r\n        def callback(msgs, exc):\r\n            # send full message to external store or log.\r\n            raise exc\r\n\r\n        producer = topic.get_producer(\r\n                delivery_reports=True,\r\n                producer_exception_callback=callback\r\n        )\r\n```\r\n\r\nThe general issue is currently there is a warning level log, but no easy way to rectify.\r\n\r\nThird, because the producer bundles the messages into a messageset. Does kafka reject the entire messageset because it is too large even though the underlying messages are small enough? \r\n\r\n","created_at":"2016-03-04T19:05:26Z","updated_at":"2016-03-19T04:13:22Z","closed_at":null,"merged_at":null,"merge_commit_sha":"32b89f1abf3c0bbce294f2029c8aa5556eb019f6","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":null,"commits_url":"https://api.github.com/repos/Parsely/pykafka/pulls/480/commits","review_comments_url":"https://api.github.com/repos/Parsely/pykafka/pulls/480/comments","review_comment_url":"https://api.github.com/repos/Parsely/pykafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/Parsely/pykafka/issues/480/comments","statuses_url":"https://api.github.com/repos/Parsely/pykafka/statuses/9c81855bb99ccf24437cd6dcdac90e729d63252f","head":{"label":"jofusa:jdennison/producer-callback","ref":"jdennison/producer-callback","sha":"9c81855bb99ccf24437cd6dcdac90e729d63252f","user":{"login":"jofusa","id":800669,"avatar_url":"https://avatars.githubusercontent.com/u/800669?v=3","gravatar_id":"","url":"https://api.github.com/users/jofusa","html_url":"https://github.com/jofusa","followers_url":"https://api.github.com/users/jofusa/followers","following_url":"https://api.github.com/users/jofusa/following{/other_user}","gists_url":"https://api.github.com/users/jofusa/gists{/gist_id}","starred_url":"https://api.github.com/users/jofusa/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jofusa/subscriptions","organizations_url":"https://api.github.com/users/jofusa/orgs","repos_url":"https://api.github.com/users/jofusa/repos","events_url":"https://api.github.com/users/jofusa/events{/privacy}","received_events_url":"https://api.github.com/users/jofusa/received_events","type":"User","site_admin":false},"repo":{"id":42269842,"name":"pykafka","full_name":"jofusa/pykafka","owner":{"login":"jofusa","id":800669,"avatar_url":"https://avatars.githubusercontent.com/u/800669?v=3","gravatar_id":"","url":"https://api.github.com/users/jofusa","html_url":"https://github.com/jofusa","followers_url":"https://api.github.com/users/jofusa/followers","following_url":"https://api.github.com/users/jofusa/following{/other_user}","gists_url":"https://api.github.com/users/jofusa/gists{/gist_id}","starred_url":"https://api.github.com/users/jofusa/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jofusa/subscriptions","organizations_url":"https://api.github.com/users/jofusa/orgs","repos_url":"https://api.github.com/users/jofusa/repos","events_url":"https://api.github.com/users/jofusa/events{/privacy}","received_events_url":"https://api.github.com/users/jofusa/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/jofusa/pykafka","description":"Kafka client for Python","fork":true,"url":"https://api.github.com/repos/jofusa/pykafka","forks_url":"https://api.github.com/repos/jofusa/pykafka/forks","keys_url":"https://api.github.com/repos/jofusa/pykafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/jofusa/pykafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/jofusa/pykafka/teams","hooks_url":"https://api.github.com/repos/jofusa/pykafka/hooks","issue_events_url":"https://api.github.com/repos/jofusa/pykafka/issues/events{/number}","events_url":"https://api.github.com/repos/jofusa/pykafka/events","assignees_url":"https://api.github.com/repos/jofusa/pykafka/assignees{/user}","branches_url":"https://api.github.com/repos/jofusa/pykafka/branches{/branch}","tags_url":"https://api.github.com/repos/jofusa/pykafka/tags","blobs_url":"https://api.github.com/repos/jofusa/pykafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/jofusa/pykafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/jofusa/pykafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/jofusa/pykafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/jofusa/pykafka/statuses/{sha}","languages_url":"https://api.github.com/repos/jofusa/pykafka/languages","stargazers_url":"https://api.github.com/repos/jofusa/pykafka/stargazers","contributors_url":"https://api.github.com/repos/jofusa/pykafka/contributors","subscribers_url":"https://api.github.com/repos/jofusa/pykafka/subscribers","subscription_url":"https://api.github.com/repos/jofusa/pykafka/subscription","commits_url":"https://api.github.com/repos/jofusa/pykafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/jofusa/pykafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/jofusa/pykafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/jofusa/pykafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/jofusa/pykafka/contents/{+path}","compare_url":"https://api.github.com/repos/jofusa/pykafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/jofusa/pykafka/merges","archive_url":"https://api.github.com/repos/jofusa/pykafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/jofusa/pykafka/downloads","issues_url":"https://api.github.com/repos/jofusa/pykafka/issues{/number}","pulls_url":"https://api.github.com/repos/jofusa/pykafka/pulls{/number}","milestones_url":"https://api.github.com/repos/jofusa/pykafka/milestones{/number}","notifications_url":"https://api.github.com/repos/jofusa/pykafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/jofusa/pykafka/labels{/name}","releases_url":"https://api.github.com/repos/jofusa/pykafka/releases{/id}","deployments_url":"https://api.github.com/repos/jofusa/pykafka/deployments","created_at":"2015-09-10T20:46:25Z","updated_at":"2015-09-10T20:46:27Z","pushed_at":"2016-03-19T02:22:11Z","git_url":"git://github.com/jofusa/pykafka.git","ssh_url":"git@github.com:jofusa/pykafka.git","clone_url":"https://github.com/jofusa/pykafka.git","svn_url":"https://github.com/jofusa/pykafka","homepage":"http://pykafka.readthedocs.org/","size":2037,"stargazers_count":0,"watchers_count":0,"language":"Python","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":"master"}},"base":{"label":"Parsely:master","ref":"master","sha":"41343b9d0ff969f654b061a627858760d3bc866b","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}","deployments_url":"https://api.github.com/repos/Parsely/pykafka/deployments","created_at":"2012-07-10T18:02:21Z","updated_at":"2016-03-18T20:14:32Z","pushed_at":"2016-03-19T02:22:11Z","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":2281,"stargazers_count":275,"watchers_count":275,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":91,"mirror_url":null,"open_issues_count":43,"forks":91,"open_issues":43,"watchers":275,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/480"},"html":{"href":"https://github.com/Parsely/pykafka/pull/480"},"issue":{"href":"https://api.github.com/repos/Parsely/pykafka/issues/480"},"comments":{"href":"https://api.github.com/repos/Parsely/pykafka/issues/480/comments"},"review_comments":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/480/comments"},"review_comment":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/Parsely/pykafka/pulls/480/commits"},"statuses":{"href":"https://api.github.com/repos/Parsely/pykafka/statuses/9c81855bb99ccf24437cd6dcdac90e729d63252f"}}}}