{"action":"created","comment":{"url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/comments/48318408","id":48318408,"diff_hunk":"@@ -91,18 +106,91 @@ KafkaProducer.prototype.connect = function connect(onConnect) {\n \n KafkaProducer.prototype.produce = function produce(topic, message, timeStamp, callback) {\n     var self = this;\n+\n     if (self.restClient) {\n-        var produceMessage = self.getProduceMessage(topic, message, timeStamp, 'binary');\n-        self.restClient.produce(produceMessage, function handleResponse(err, res) {\n-            if (callback) {\n-                callback(err, res);\n-            }\n-        });\n+        if (self.batching) {\n+            self.batch(topic, message, timeStamp, callback);\n+        } else {\n+            var produceMessage = self.getProduceMessage(topic, message, timeStamp, 'binary');\n+            self.restClient.produce(produceMessage, function handleResponse(err, res) {\n+                if (callback) {\n+                    callback(err, res);\n+                }\n+            });\n+        }\n     } else if (callback) {\n         callback(new Error('Kafka Rest Client is not initialized!'));\n     }\n };\n \n+// Add message to topic's BatchMessage in cache or add new topic to cache\n+// Flush topic's BatchMessage on flushCycleSecs interval or if greater than maxBatchSizeBytes\n+KafkaProducer.prototype.batch = function batch(topic, message, timeStamp, callback) {\n+    var self = this;\n+    var currTime = new Date().getTime();\n+    if (topic in self.topicToBatchQueue) {\n+        var queue = self.topicToBatchQueue[topic];\n+        if ((queue.messageBatch.sizeBytes + message.length) > self.maxBatchSizeBytes ||\n+                (currTime - queue.lastFlush) > self.flushCycleSecs * 1000) {\n+            var produceMessage = self.getProduceMessage(topic,\n+                queue.messageBatch.getBatchedMessage(),\n+                timeStamp,\n+                'batch');\n+            self.restClient.produce(produceMessage, function handleResponse(err, res) {\n+                if (callback) {\n+                    callback(err, res);\n+                }\n+            });\n+\n+            queue.messageBatch = new MessageBatch(topic);\n+            queue.lastFlush = new Date().getTime();\n+        }\n+\n+        queue.messageBatch.addMessage(message);\n+    } else {\n+        var messageBatch = new MessageBatch(topic);\n+        messageBatch.addMessage(message);\n+\n+        var batchQueue = {\n+            messageBatch: messageBatch,\n+            lastFlush: currTime\n+        };\n+\n+        self.topicToBatchQueue[topic] = batchQueue;\n+    }\n+    callback();\n+};\n+\n+KafkaProducer.prototype.flushEntireCache = function flushEntireCache(callback) {\n+    var self = this;\n+\n+    var handleResponse = function handleResponse(err, res) {\n+        if (callback) {\n+            callback(err, res);\n+        }\n+    };\n+\n+    for (var topic in self.topicToBatchQueue) {\n+        if (self.topicToBatchQueue.hasOwnProperty[topic]) {\n+            var queue = self.topicToBatchQueue[topic];\n+            var currTime = new Date().getTime();\n+\n+            if ((currTime - queue.lastFlush) > self.flushCycleSecs * 1000 &&\n+                    queue.messageBatch.numMessages > 0) {","path":"lib/kafka_producer.js","position":111,"original_position":112,"commit_id":"52132a9aa198574af85772bf14a27d709511ac8e","original_commit_id":"f7dc6b3ae74f2a807e5adf3599afaac72a12ad07","user":{"login":"qiliu-stanford","id":9031072,"avatar_url":"https://avatars.githubusercontent.com/u/9031072?v=3","gravatar_id":"","url":"https://api.github.com/users/qiliu-stanford","html_url":"https://github.com/qiliu-stanford","followers_url":"https://api.github.com/users/qiliu-stanford/followers","following_url":"https://api.github.com/users/qiliu-stanford/following{/other_user}","gists_url":"https://api.github.com/users/qiliu-stanford/gists{/gist_id}","starred_url":"https://api.github.com/users/qiliu-stanford/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/qiliu-stanford/subscriptions","organizations_url":"https://api.github.com/users/qiliu-stanford/orgs","repos_url":"https://api.github.com/users/qiliu-stanford/repos","events_url":"https://api.github.com/users/qiliu-stanford/events{/privacy}","received_events_url":"https://api.github.com/users/qiliu-stanford/received_events","type":"User","site_admin":false},"body":"If it's the flushing on close, shouldn't you flush regardless of the lastFlush if numMessages > 0","created_at":"2015-12-23T02:40:23Z","updated_at":"2015-12-23T02:40:23Z","html_url":"https://github.com/uber-common/node-kafka-rest-client/pull/12#discussion_r48318408","pull_request_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12","_links":{"self":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/comments/48318408"},"html":{"href":"https://github.com/uber-common/node-kafka-rest-client/pull/12#discussion_r48318408"},"pull_request":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12"}}},"pull_request":{"url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12","id":54315893,"html_url":"https://github.com/uber-common/node-kafka-rest-client/pull/12","diff_url":"https://github.com/uber-common/node-kafka-rest-client/pull/12.diff","patch_url":"https://github.com/uber-common/node-kafka-rest-client/pull/12.patch","issue_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/12","number":12,"state":"open","locked":false,"title":"add batching to kafka producer","user":{"login":"bwwinthehouse","id":5456704,"avatar_url":"https://avatars.githubusercontent.com/u/5456704?v=3","gravatar_id":"","url":"https://api.github.com/users/bwwinthehouse","html_url":"https://github.com/bwwinthehouse","followers_url":"https://api.github.com/users/bwwinthehouse/followers","following_url":"https://api.github.com/users/bwwinthehouse/following{/other_user}","gists_url":"https://api.github.com/users/bwwinthehouse/gists{/gist_id}","starred_url":"https://api.github.com/users/bwwinthehouse/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bwwinthehouse/subscriptions","organizations_url":"https://api.github.com/users/bwwinthehouse/orgs","repos_url":"https://api.github.com/users/bwwinthehouse/repos","events_url":"https://api.github.com/users/bwwinthehouse/events{/privacy}","received_events_url":"https://api.github.com/users/bwwinthehouse/received_events","type":"User","site_admin":false},"body":"Example use: https://code.uberinternal.com/diffusion/DANODE/browse/master/rest_client_example.js","created_at":"2015-12-21T23:29:56Z","updated_at":"2015-12-23T02:40:23Z","closed_at":null,"merged_at":null,"merge_commit_sha":"cb9e99754a756430a9ae03462471511380e18466","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12/commits","review_comments_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12/comments","review_comment_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/comments{/number}","comments_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/12/comments","statuses_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/statuses/52132a9aa198574af85772bf14a27d709511ac8e","head":{"label":"uber-common:batch-in-kafka-producer","ref":"batch-in-kafka-producer","sha":"52132a9aa198574af85772bf14a27d709511ac8e","user":{"login":"uber-common","id":14812045,"avatar_url":"https://avatars.githubusercontent.com/u/14812045?v=3","gravatar_id":"","url":"https://api.github.com/users/uber-common","html_url":"https://github.com/uber-common","followers_url":"https://api.github.com/users/uber-common/followers","following_url":"https://api.github.com/users/uber-common/following{/other_user}","gists_url":"https://api.github.com/users/uber-common/gists{/gist_id}","starred_url":"https://api.github.com/users/uber-common/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uber-common/subscriptions","organizations_url":"https://api.github.com/users/uber-common/orgs","repos_url":"https://api.github.com/users/uber-common/repos","events_url":"https://api.github.com/users/uber-common/events{/privacy}","received_events_url":"https://api.github.com/users/uber-common/received_events","type":"Organization","site_admin":false},"repo":{"id":41319172,"name":"node-kafka-rest-client","full_name":"uber-common/node-kafka-rest-client","owner":{"login":"uber-common","id":14812045,"avatar_url":"https://avatars.githubusercontent.com/u/14812045?v=3","gravatar_id":"","url":"https://api.github.com/users/uber-common","html_url":"https://github.com/uber-common","followers_url":"https://api.github.com/users/uber-common/followers","following_url":"https://api.github.com/users/uber-common/following{/other_user}","gists_url":"https://api.github.com/users/uber-common/gists{/gist_id}","starred_url":"https://api.github.com/users/uber-common/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uber-common/subscriptions","organizations_url":"https://api.github.com/users/uber-common/orgs","repos_url":"https://api.github.com/users/uber-common/repos","events_url":"https://api.github.com/users/uber-common/events{/privacy}","received_events_url":"https://api.github.com/users/uber-common/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/uber-common/node-kafka-rest-client","description":"A REST client for producing messages to Kafka","fork":false,"url":"https://api.github.com/repos/uber-common/node-kafka-rest-client","forks_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/forks","keys_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/keys{/key_id}","collaborators_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/teams","hooks_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/hooks","issue_events_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/events{/number}","events_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/events","assignees_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/assignees{/user}","branches_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/branches{/branch}","tags_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/tags","blobs_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/refs{/sha}","trees_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/trees{/sha}","statuses_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/statuses/{sha}","languages_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/languages","stargazers_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/stargazers","contributors_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/contributors","subscribers_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/subscribers","subscription_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/subscription","commits_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/commits{/sha}","git_commits_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/commits{/sha}","comments_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/comments{/number}","issue_comment_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/comments{/number}","contents_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/contents/{+path}","compare_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/compare/{base}...{head}","merges_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/merges","archive_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/downloads","issues_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues{/number}","pulls_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls{/number}","milestones_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/milestones{/number}","notifications_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/labels{/name}","releases_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/releases{/id}","created_at":"2015-08-24T18:12:37Z","updated_at":"2015-10-13T20:05:14Z","pushed_at":"2015-12-23T01:34:28Z","git_url":"git://github.com/uber-common/node-kafka-rest-client.git","ssh_url":"git@github.com:uber-common/node-kafka-rest-client.git","clone_url":"https://github.com/uber-common/node-kafka-rest-client.git","svn_url":"https://github.com/uber-common/node-kafka-rest-client","homepage":"http://www.uber.com","size":62,"stargazers_count":0,"watchers_count":0,"language":"JavaScript","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":1,"forks":1,"open_issues":1,"watchers":0,"default_branch":"master"}},"base":{"label":"uber-common:master","ref":"master","sha":"d398245fad023064ebd60c9ce4ef20ee23b8ce1c","user":{"login":"uber-common","id":14812045,"avatar_url":"https://avatars.githubusercontent.com/u/14812045?v=3","gravatar_id":"","url":"https://api.github.com/users/uber-common","html_url":"https://github.com/uber-common","followers_url":"https://api.github.com/users/uber-common/followers","following_url":"https://api.github.com/users/uber-common/following{/other_user}","gists_url":"https://api.github.com/users/uber-common/gists{/gist_id}","starred_url":"https://api.github.com/users/uber-common/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uber-common/subscriptions","organizations_url":"https://api.github.com/users/uber-common/orgs","repos_url":"https://api.github.com/users/uber-common/repos","events_url":"https://api.github.com/users/uber-common/events{/privacy}","received_events_url":"https://api.github.com/users/uber-common/received_events","type":"Organization","site_admin":false},"repo":{"id":41319172,"name":"node-kafka-rest-client","full_name":"uber-common/node-kafka-rest-client","owner":{"login":"uber-common","id":14812045,"avatar_url":"https://avatars.githubusercontent.com/u/14812045?v=3","gravatar_id":"","url":"https://api.github.com/users/uber-common","html_url":"https://github.com/uber-common","followers_url":"https://api.github.com/users/uber-common/followers","following_url":"https://api.github.com/users/uber-common/following{/other_user}","gists_url":"https://api.github.com/users/uber-common/gists{/gist_id}","starred_url":"https://api.github.com/users/uber-common/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/uber-common/subscriptions","organizations_url":"https://api.github.com/users/uber-common/orgs","repos_url":"https://api.github.com/users/uber-common/repos","events_url":"https://api.github.com/users/uber-common/events{/privacy}","received_events_url":"https://api.github.com/users/uber-common/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/uber-common/node-kafka-rest-client","description":"A REST client for producing messages to Kafka","fork":false,"url":"https://api.github.com/repos/uber-common/node-kafka-rest-client","forks_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/forks","keys_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/keys{/key_id}","collaborators_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/teams","hooks_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/hooks","issue_events_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/events{/number}","events_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/events","assignees_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/assignees{/user}","branches_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/branches{/branch}","tags_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/tags","blobs_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/refs{/sha}","trees_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/trees{/sha}","statuses_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/statuses/{sha}","languages_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/languages","stargazers_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/stargazers","contributors_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/contributors","subscribers_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/subscribers","subscription_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/subscription","commits_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/commits{/sha}","git_commits_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/git/commits{/sha}","comments_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/comments{/number}","issue_comment_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/comments{/number}","contents_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/contents/{+path}","compare_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/compare/{base}...{head}","merges_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/merges","archive_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/downloads","issues_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues{/number}","pulls_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls{/number}","milestones_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/milestones{/number}","notifications_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/labels{/name}","releases_url":"https://api.github.com/repos/uber-common/node-kafka-rest-client/releases{/id}","created_at":"2015-08-24T18:12:37Z","updated_at":"2015-10-13T20:05:14Z","pushed_at":"2015-12-23T01:34:28Z","git_url":"git://github.com/uber-common/node-kafka-rest-client.git","ssh_url":"git@github.com:uber-common/node-kafka-rest-client.git","clone_url":"https://github.com/uber-common/node-kafka-rest-client.git","svn_url":"https://github.com/uber-common/node-kafka-rest-client","homepage":"http://www.uber.com","size":62,"stargazers_count":0,"watchers_count":0,"language":"JavaScript","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":1,"mirror_url":null,"open_issues_count":1,"forks":1,"open_issues":1,"watchers":0,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12"},"html":{"href":"https://github.com/uber-common/node-kafka-rest-client/pull/12"},"issue":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/12"},"comments":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/issues/12/comments"},"review_comments":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12/comments"},"review_comment":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/pulls/12/commits"},"statuses":{"href":"https://api.github.com/repos/uber-common/node-kafka-rest-client/statuses/52132a9aa198574af85772bf14a27d709511ac8e"}}}}