{"action":"created","comment":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/comments/71178308","id":71178308,"diff_hunk":"@@ -0,0 +1,222 @@\n+'use strict';\n+var util = require('util');\n+var EventEmitter = require('events');\n+var sinon = require('sinon');\n+\n+var libPath = process.env['KAFKA_COV'] ? '../lib-cov/' : '../lib/';\n+\n+// var Client = require(libPath + 'client');\n+var HighLevelConsumer = require(libPath + 'highLevelConsumer');\n+\n+function FakeZookeeper () {\n+  EventEmitter.call(this);\n+\n+  this.checkPartitionOwnership = function (id, groupId, topic, partition, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.deletePartitionOwnership = function (groupId, topic, partition, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.addPartitionOwnership = function (id, groupId, topic, partition, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.registerConsumer = function (groupId, id, payloads, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.listConsumers = function (groupId) {};\n+  this.getConsumersPerTopic = function (groupId, cb) {\n+    var consumerTopicMap = {};\n+    consumerTopicMap[groupId] = [ 'fake-topic' ];\n+    var topicConsumerMap = { 'fake-topic': [ groupId ] };\n+    var topicPartitionMap = { 'fake-topic': [ '0', '1', '2' ] };\n+    var map = {\n+      consumerTopicMap: consumerTopicMap,\n+      topicConsumerMap: topicConsumerMap,\n+      topicPartitionMap: topicPartitionMap\n+    };\n+\n+    setTimeout(\n+      cb.bind(null, null, map),\n+      0\n+    );\n+  };\n+}\n+util.inherits(FakeZookeeper, EventEmitter);\n+\n+function FakeClient () {\n+  EventEmitter.call(this);\n+\n+  this.zk = new FakeZookeeper();\n+\n+  this.topicExists = function (topics, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.refreshMetadata = function (topicNames, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.sendOffsetCommitRequest = function (groupId, commits, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.sendFetchRequest = function (consumer, payloads, fetchMaxWaitMs, fetchMinBytes, maxTickMessages) {};\n+  this.sendOffsetFetchRequest = function (groupId, payloads, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.sendOffsetRequest = function (payloads, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.addTopics = function (topics, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.removeTopicMetadata = function (topics, cb) {\n+    setTimeout(cb, 0);\n+  };\n+  this.close = function (cb) {\n+    setTimeout(cb, 0);\n+  };\n+}\n+util.inherits(FakeClient, EventEmitter);\n+\n+describe('HighLevelConsumer', function () {\n+  describe('rebalance', function () {\n+    var client,\n+      highLevelConsumer,\n+      sandbox;\n+\n+    beforeEach(function () {\n+      client = new FakeClient();\n+\n+      highLevelConsumer = new HighLevelConsumer(\n+        client,\n+        [ {topic: 'fake-topic'} ]\n+      );\n+\n+      clearTimeout(highLevelConsumer.checkPartitionOwnershipInterval);\n+\n+      sandbox = sinon.sandbox.create();\n+    });\n+\n+    afterEach(function () {\n+      highLevelConsumer.close(function () {});\n+      sandbox.restore();\n+      client = null;\n+      highLevelConsumer = null;\n+    });\n+\n+    it('should emit rebalanced event and clear rebalancing flag only after offsets are updated', function (done) {\n+      client.emit('ready');\n+\n+      highLevelConsumer.on('registered', function () {\n+        var sendOffsetFetchRequestCalled = false;\n+        var refreshMetadataCalled = false;\n+\n+        // verify rebalancing is false until rebalance finishes\n+        var refreshMetadata = client.refreshMetadata;\n+        client.refreshMetadata = function () {","path":"test/test.highlevelConsumer.js","position":114,"original_position":114,"commit_id":"e3158549ff7b9e22ba499dbb838e04226f3fbcc8","original_commit_id":"e3158549ff7b9e22ba499dbb838e04226f3fbcc8","user":{"login":"hyperlink","id":133309,"avatar_url":"https://avatars.githubusercontent.com/u/133309?v=3","gravatar_id":"","url":"https://api.github.com/users/hyperlink","html_url":"https://github.com/hyperlink","followers_url":"https://api.github.com/users/hyperlink/followers","following_url":"https://api.github.com/users/hyperlink/following{/other_user}","gists_url":"https://api.github.com/users/hyperlink/gists{/gist_id}","starred_url":"https://api.github.com/users/hyperlink/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/hyperlink/subscriptions","organizations_url":"https://api.github.com/users/hyperlink/orgs","repos_url":"https://api.github.com/users/hyperlink/repos","events_url":"https://api.github.com/users/hyperlink/events{/privacy}","received_events_url":"https://api.github.com/users/hyperlink/received_events","type":"User","site_admin":false},"body":"This can be shortened by using a stub. And later you can `sinon.assert` that this stub was called.\r\n\r\n```js\r\n        sandbox.stub(client, 'sendOffsetFetchRequest', function (groupId, payloads, cb) {\r\n          highLevelConsumer.rebalancing.should.be.true;\r\n          setImmediate(cb);\r\n        });\r\n```","created_at":"2016-07-18T16:16:10Z","updated_at":"2016-07-18T16:16:10Z","html_url":"https://github.com/SOHU-Co/kafka-node/pull/402#discussion_r71178308","pull_request_url":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402","_links":{"self":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/comments/71178308"},"html":{"href":"https://github.com/SOHU-Co/kafka-node/pull/402#discussion_r71178308"},"pull_request":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402"}}},"pull_request":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402","id":77546647,"html_url":"https://github.com/SOHU-Co/kafka-node/pull/402","diff_url":"https://github.com/SOHU-Co/kafka-node/pull/402.diff","patch_url":"https://github.com/SOHU-Co/kafka-node/pull/402.patch","issue_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/402","number":402,"state":"open","locked":false,"title":"Prevent fetch before offset are updated after high level consumer rebalance","user":{"login":"snatesan","id":1195258,"avatar_url":"https://avatars.githubusercontent.com/u/1195258?v=3","gravatar_id":"","url":"https://api.github.com/users/snatesan","html_url":"https://github.com/snatesan","followers_url":"https://api.github.com/users/snatesan/followers","following_url":"https://api.github.com/users/snatesan/following{/other_user}","gists_url":"https://api.github.com/users/snatesan/gists{/gist_id}","starred_url":"https://api.github.com/users/snatesan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/snatesan/subscriptions","organizations_url":"https://api.github.com/users/snatesan/orgs","repos_url":"https://api.github.com/users/snatesan/repos","events_url":"https://api.github.com/users/snatesan/events{/privacy}","received_events_url":"https://api.github.com/users/snatesan/received_events","type":"User","site_admin":false},"body":"If there is done event fired on high level consumer \r\nafter \"rebalanced\" event \r\nbut before \"updateOffsets\" method in \"rebalanced\" event handler of HighLevelConsumer being fulfilled \r\nthen that will trigger fetching all messages from zookeeper for the partition requested.\r\nThis is happening because at that stage owned partitions of HighLevelConsumer is set to 0.\r\nThis PR make sure \"rebalancing\" flag is set to \"false\" only after offsets are updated with the values from zookeeper and hence any \"fetch\" call before that will be ignored.\r\n\r\nThis issue can happen when rebalance occurs frequently in short time due to various reasons like network interruptions or addition or removal of a consumer with same group id.","created_at":"2016-07-15T00:17:27Z","updated_at":"2016-07-18T16:16:10Z","closed_at":null,"merged_at":null,"merge_commit_sha":"a199b0ec8724b375a77ab9263bd6c64366c81c56","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402/commits","review_comments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402/comments","review_comment_url":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/comments{/number}","comments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/402/comments","statuses_url":"https://api.github.com/repos/SOHU-Co/kafka-node/statuses/e3158549ff7b9e22ba499dbb838e04226f3fbcc8","head":{"label":"snatesan:master","ref":"master","sha":"e3158549ff7b9e22ba499dbb838e04226f3fbcc8","user":{"login":"snatesan","id":1195258,"avatar_url":"https://avatars.githubusercontent.com/u/1195258?v=3","gravatar_id":"","url":"https://api.github.com/users/snatesan","html_url":"https://github.com/snatesan","followers_url":"https://api.github.com/users/snatesan/followers","following_url":"https://api.github.com/users/snatesan/following{/other_user}","gists_url":"https://api.github.com/users/snatesan/gists{/gist_id}","starred_url":"https://api.github.com/users/snatesan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/snatesan/subscriptions","organizations_url":"https://api.github.com/users/snatesan/orgs","repos_url":"https://api.github.com/users/snatesan/repos","events_url":"https://api.github.com/users/snatesan/events{/privacy}","received_events_url":"https://api.github.com/users/snatesan/received_events","type":"User","site_admin":false},"repo":{"id":62910197,"name":"kafka-node","full_name":"snatesan/kafka-node","owner":{"login":"snatesan","id":1195258,"avatar_url":"https://avatars.githubusercontent.com/u/1195258?v=3","gravatar_id":"","url":"https://api.github.com/users/snatesan","html_url":"https://github.com/snatesan","followers_url":"https://api.github.com/users/snatesan/followers","following_url":"https://api.github.com/users/snatesan/following{/other_user}","gists_url":"https://api.github.com/users/snatesan/gists{/gist_id}","starred_url":"https://api.github.com/users/snatesan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/snatesan/subscriptions","organizations_url":"https://api.github.com/users/snatesan/orgs","repos_url":"https://api.github.com/users/snatesan/repos","events_url":"https://api.github.com/users/snatesan/events{/privacy}","received_events_url":"https://api.github.com/users/snatesan/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/snatesan/kafka-node","description":"Node.js client for Apache Kafka 0.8","fork":true,"url":"https://api.github.com/repos/snatesan/kafka-node","forks_url":"https://api.github.com/repos/snatesan/kafka-node/forks","keys_url":"https://api.github.com/repos/snatesan/kafka-node/keys{/key_id}","collaborators_url":"https://api.github.com/repos/snatesan/kafka-node/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/snatesan/kafka-node/teams","hooks_url":"https://api.github.com/repos/snatesan/kafka-node/hooks","issue_events_url":"https://api.github.com/repos/snatesan/kafka-node/issues/events{/number}","events_url":"https://api.github.com/repos/snatesan/kafka-node/events","assignees_url":"https://api.github.com/repos/snatesan/kafka-node/assignees{/user}","branches_url":"https://api.github.com/repos/snatesan/kafka-node/branches{/branch}","tags_url":"https://api.github.com/repos/snatesan/kafka-node/tags","blobs_url":"https://api.github.com/repos/snatesan/kafka-node/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/snatesan/kafka-node/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/snatesan/kafka-node/git/refs{/sha}","trees_url":"https://api.github.com/repos/snatesan/kafka-node/git/trees{/sha}","statuses_url":"https://api.github.com/repos/snatesan/kafka-node/statuses/{sha}","languages_url":"https://api.github.com/repos/snatesan/kafka-node/languages","stargazers_url":"https://api.github.com/repos/snatesan/kafka-node/stargazers","contributors_url":"https://api.github.com/repos/snatesan/kafka-node/contributors","subscribers_url":"https://api.github.com/repos/snatesan/kafka-node/subscribers","subscription_url":"https://api.github.com/repos/snatesan/kafka-node/subscription","commits_url":"https://api.github.com/repos/snatesan/kafka-node/commits{/sha}","git_commits_url":"https://api.github.com/repos/snatesan/kafka-node/git/commits{/sha}","comments_url":"https://api.github.com/repos/snatesan/kafka-node/comments{/number}","issue_comment_url":"https://api.github.com/repos/snatesan/kafka-node/issues/comments{/number}","contents_url":"https://api.github.com/repos/snatesan/kafka-node/contents/{+path}","compare_url":"https://api.github.com/repos/snatesan/kafka-node/compare/{base}...{head}","merges_url":"https://api.github.com/repos/snatesan/kafka-node/merges","archive_url":"https://api.github.com/repos/snatesan/kafka-node/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/snatesan/kafka-node/downloads","issues_url":"https://api.github.com/repos/snatesan/kafka-node/issues{/number}","pulls_url":"https://api.github.com/repos/snatesan/kafka-node/pulls{/number}","milestones_url":"https://api.github.com/repos/snatesan/kafka-node/milestones{/number}","notifications_url":"https://api.github.com/repos/snatesan/kafka-node/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/snatesan/kafka-node/labels{/name}","releases_url":"https://api.github.com/repos/snatesan/kafka-node/releases{/id}","deployments_url":"https://api.github.com/repos/snatesan/kafka-node/deployments","created_at":"2016-07-08T18:47:27Z","updated_at":"2016-07-08T18:47:28Z","pushed_at":"2016-07-16T01:22:23Z","git_url":"git://github.com/snatesan/kafka-node.git","ssh_url":"git@github.com:snatesan/kafka-node.git","clone_url":"https://github.com/snatesan/kafka-node.git","svn_url":"https://github.com/snatesan/kafka-node","homepage":"","size":622,"stargazers_count":0,"watchers_count":0,"language":"JavaScript","has_issues":false,"has_downloads":true,"has_wiki":true,"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":"SOHU-Co:master","ref":"master","sha":"98fca8c23f62ed5d891b48325316e052a2828b7e","user":{"login":"SOHU-Co","id":5337985,"avatar_url":"https://avatars.githubusercontent.com/u/5337985?v=3","gravatar_id":"","url":"https://api.github.com/users/SOHU-Co","html_url":"https://github.com/SOHU-Co","followers_url":"https://api.github.com/users/SOHU-Co/followers","following_url":"https://api.github.com/users/SOHU-Co/following{/other_user}","gists_url":"https://api.github.com/users/SOHU-Co/gists{/gist_id}","starred_url":"https://api.github.com/users/SOHU-Co/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SOHU-Co/subscriptions","organizations_url":"https://api.github.com/users/SOHU-Co/orgs","repos_url":"https://api.github.com/users/SOHU-Co/repos","events_url":"https://api.github.com/users/SOHU-Co/events{/privacy}","received_events_url":"https://api.github.com/users/SOHU-Co/received_events","type":"Organization","site_admin":false},"repo":{"id":13792721,"name":"kafka-node","full_name":"SOHU-Co/kafka-node","owner":{"login":"SOHU-Co","id":5337985,"avatar_url":"https://avatars.githubusercontent.com/u/5337985?v=3","gravatar_id":"","url":"https://api.github.com/users/SOHU-Co","html_url":"https://github.com/SOHU-Co","followers_url":"https://api.github.com/users/SOHU-Co/followers","following_url":"https://api.github.com/users/SOHU-Co/following{/other_user}","gists_url":"https://api.github.com/users/SOHU-Co/gists{/gist_id}","starred_url":"https://api.github.com/users/SOHU-Co/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SOHU-Co/subscriptions","organizations_url":"https://api.github.com/users/SOHU-Co/orgs","repos_url":"https://api.github.com/users/SOHU-Co/repos","events_url":"https://api.github.com/users/SOHU-Co/events{/privacy}","received_events_url":"https://api.github.com/users/SOHU-Co/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/SOHU-Co/kafka-node","description":"Node.js client for Apache Kafka 0.8","fork":false,"url":"https://api.github.com/repos/SOHU-Co/kafka-node","forks_url":"https://api.github.com/repos/SOHU-Co/kafka-node/forks","keys_url":"https://api.github.com/repos/SOHU-Co/kafka-node/keys{/key_id}","collaborators_url":"https://api.github.com/repos/SOHU-Co/kafka-node/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/SOHU-Co/kafka-node/teams","hooks_url":"https://api.github.com/repos/SOHU-Co/kafka-node/hooks","issue_events_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/events{/number}","events_url":"https://api.github.com/repos/SOHU-Co/kafka-node/events","assignees_url":"https://api.github.com/repos/SOHU-Co/kafka-node/assignees{/user}","branches_url":"https://api.github.com/repos/SOHU-Co/kafka-node/branches{/branch}","tags_url":"https://api.github.com/repos/SOHU-Co/kafka-node/tags","blobs_url":"https://api.github.com/repos/SOHU-Co/kafka-node/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/SOHU-Co/kafka-node/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/SOHU-Co/kafka-node/git/refs{/sha}","trees_url":"https://api.github.com/repos/SOHU-Co/kafka-node/git/trees{/sha}","statuses_url":"https://api.github.com/repos/SOHU-Co/kafka-node/statuses/{sha}","languages_url":"https://api.github.com/repos/SOHU-Co/kafka-node/languages","stargazers_url":"https://api.github.com/repos/SOHU-Co/kafka-node/stargazers","contributors_url":"https://api.github.com/repos/SOHU-Co/kafka-node/contributors","subscribers_url":"https://api.github.com/repos/SOHU-Co/kafka-node/subscribers","subscription_url":"https://api.github.com/repos/SOHU-Co/kafka-node/subscription","commits_url":"https://api.github.com/repos/SOHU-Co/kafka-node/commits{/sha}","git_commits_url":"https://api.github.com/repos/SOHU-Co/kafka-node/git/commits{/sha}","comments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/comments{/number}","issue_comment_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/comments{/number}","contents_url":"https://api.github.com/repos/SOHU-Co/kafka-node/contents/{+path}","compare_url":"https://api.github.com/repos/SOHU-Co/kafka-node/compare/{base}...{head}","merges_url":"https://api.github.com/repos/SOHU-Co/kafka-node/merges","archive_url":"https://api.github.com/repos/SOHU-Co/kafka-node/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/SOHU-Co/kafka-node/downloads","issues_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues{/number}","pulls_url":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls{/number}","milestones_url":"https://api.github.com/repos/SOHU-Co/kafka-node/milestones{/number}","notifications_url":"https://api.github.com/repos/SOHU-Co/kafka-node/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/SOHU-Co/kafka-node/labels{/name}","releases_url":"https://api.github.com/repos/SOHU-Co/kafka-node/releases{/id}","deployments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/deployments","created_at":"2013-10-23T03:34:05Z","updated_at":"2016-07-18T14:02:39Z","pushed_at":"2016-07-16T01:22:23Z","git_url":"git://github.com/SOHU-Co/kafka-node.git","ssh_url":"git@github.com:SOHU-Co/kafka-node.git","clone_url":"https://github.com/SOHU-Co/kafka-node.git","svn_url":"https://github.com/SOHU-Co/kafka-node","homepage":"","size":985,"stargazers_count":507,"watchers_count":507,"language":"JavaScript","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":230,"mirror_url":null,"open_issues_count":115,"forks":230,"open_issues":115,"watchers":507,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402"},"html":{"href":"https://github.com/SOHU-Co/kafka-node/pull/402"},"issue":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/402"},"comments":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/402/comments"},"review_comments":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402/comments"},"review_comment":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/pulls/402/commits"},"statuses":{"href":"https://api.github.com/repos/SOHU-Co/kafka-node/statuses/e3158549ff7b9e22ba499dbb838e04226f3fbcc8"}}}}
{"actor":{"display_login":"hyperlink"}}