{"action":"created","issue":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/418","repository_url":"https://api.github.com/repos/SOHU-Co/kafka-node","labels_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/418/labels{/name}","comments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/418/comments","events_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/418/events","html_url":"https://github.com/SOHU-Co/kafka-node/issues/418","id":168256503,"number":418,"title":"HighLevelConsumer rebalance logic for offset not works perfectly in v0.5.1","user":{"login":"1more","id":8219615,"avatar_url":"https://avatars.githubusercontent.com/u/8219615?v=3","gravatar_id":"","url":"https://api.github.com/users/1more","html_url":"https://github.com/1more","followers_url":"https://api.github.com/users/1more/followers","following_url":"https://api.github.com/users/1more/following{/other_user}","gists_url":"https://api.github.com/users/1more/gists{/gist_id}","starred_url":"https://api.github.com/users/1more/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/1more/subscriptions","organizations_url":"https://api.github.com/users/1more/orgs","repos_url":"https://api.github.com/users/1more/repos","events_url":"https://api.github.com/users/1more/events{/privacy}","received_events_url":"https://api.github.com/users/1more/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":2,"created_at":"2016-07-29T06:00:23Z","updated_at":"2016-08-02T03:12:12Z","closed_at":null,"body":"Hi,\r\nI'm using kafka-node and thanks to v0.5.1 update which resolves [this](https://github.com/SOHU-Co/kafka-node/pull/402) issue.\r\nHowever, even after update to v0.5.1, rebalance logic seems have some problem.\r\nMy situation is,\r\n\r\n- Kafka server\r\n  - 6 partiton for an topic (Kafka version is 0.9.x)\r\n- Kafka client(Application server)\r\n  - 4 server, and each server has 2 node instances(using pm2) - total 8 node server instances\r\n\r\n- code for kafka client connection\r\n```javascript\r\nlet client = new kafka.Client(config.kafka.hosts)\r\nlet consumer = new kafka.HighLevelConsumer(client, [{\r\n   topic: 'someTopic'\r\n}, {\r\n   topic: 'otherTopic'\r\n}], {\r\n    groupId: 'group-id-production',\r\n    autoCommit: true,\r\n    autoCommitIntervalMs: 3000\r\n})\r\n```\r\n\r\nEvery time I run the servers, one(or two) of my kafka partition does not consumed at all.\r\n(I'm not sure it was happened in previous version(I used v0.3.x before update))\r\n\r\nIs there anyone having similar issue with me? or any solution about it?"},"comment":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/comments/236783024","html_url":"https://github.com/SOHU-Co/kafka-node/issues/418#issuecomment-236783024","issue_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/418","id":236783024,"user":{"login":"1more","id":8219615,"avatar_url":"https://avatars.githubusercontent.com/u/8219615?v=3","gravatar_id":"","url":"https://api.github.com/users/1more","html_url":"https://github.com/1more","followers_url":"https://api.github.com/users/1more/followers","following_url":"https://api.github.com/users/1more/following{/other_user}","gists_url":"https://api.github.com/users/1more/gists{/gist_id}","starred_url":"https://api.github.com/users/1more/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/1more/subscriptions","organizations_url":"https://api.github.com/users/1more/orgs","repos_url":"https://api.github.com/users/1more/repos","events_url":"https://api.github.com/users/1more/events{/privacy}","received_events_url":"https://api.github.com/users/1more/received_events","type":"User","site_admin":false},"created_at":"2016-08-02T03:12:12Z","updated_at":"2016-08-02T03:12:12Z","body":"- I think the problem was error handler logic for consumer instance in my code\r\n\r\n - my code was\r\n```javascript\r\nmodule.exports = function (app) {\r\n  let io = app.get('socketio')\r\n  let client = new kafka.Client(config.kafka.hosts)\r\n  let highLevelConsumer = initConsumer()\r\n\r\n  function initConsumer () {\r\n    let consumer = new kafka.HighLevelConsumer(client, [ {\r\n      topic: 'topic1'\r\n    }, {\r\n      topic: 'topic2'\r\n    } ], {\r\n      groupId: 'consumer-group',\r\n      autoCommit: true,\r\n      autoCommitIntervalMs: 3000\r\n    })\r\n\r\n    consumer.on('message', function (message) {\r\n      // do something\r\n    })\r\n\r\n    consumer.on('error', function (err) {\r\n      setTimeout(function () {\r\n        highLevelConsumer = initConsumer()\r\n      }, 30000)\r\n      consumer.close(true, function () {})\r\n    })\r\n    return consumer\r\n  }\r\n\r\n  process.on('SIGINT', function () {\r\n    highLevelConsumer.close(true, function () {\r\n      process.exit()\r\n    })\r\n  })\r\n}\r\n```\r\n\r\n - After debug, I found that 'NODE_EXISTS' error always occurred right after restarting server\r\n - and the consumer which throws the error not closed and remained in consumer list\r\n\r\n\r\n- So I change my codes like\r\n```javascript\r\nmodule.exports = function (app) {\r\n  let io = app.get('socketio')\r\n  let client = new kafka.Client(config.kafka.hosts)\r\n  let highLevelConsumer = initConsumer()\r\n  initConsumerEventListeners()\r\n\r\n  function initConsumer () {\r\n    let consumer = new kafka.HighLevelConsumer(client, [{\r\n      topic: 'topic1'\r\n    }, {\r\n      topic: 'topic2'\r\n    }], {\r\n      groupId: 'consumer-group',\r\n      autoCommit: true,\r\n      autoCommitIntervalMs: 3000\r\n    })\r\n    return consumer\r\n  }\r\n\r\n  function initConsumerEventListeners () {\r\n    highLevelConsumer.on('message', function (message) {\r\n      // do something\r\n    })\r\n\r\n    highLevelConsumer.on('error', function (err) {\r\n      highLevelConsumer.close(function () {\r\n        logger.info('Consumer closed inside error event handler')\r\n      })\r\n      client.close(function () {\r\n        logger.info('Client closed inside error event handler')\r\n      })\r\n      setTimeout(function () {\r\n        client = new kafka.Client(config.kafka.hosts)\r\n        highLevelConsumer = initConsumer()\r\n        initConsumerEventListeners()\r\n      }, 30000)\r\n    })\r\n  }\r\n\r\n  process.on('SIGINT', function () {\r\n    highLevelConsumer.close(true, function () {\r\n      process.exit()\r\n    })\r\n  })\r\n}\r\n```\r\n\r\nNow it works well. Thanks for answering me."}}
{"actor":{"display_login":"1more"}}