{"action":"created","issue":{"url":"https://api.github.com/repos/Parsely/pykafka/issues/538","repository_url":"https://api.github.com/repos/Parsely/pykafka","labels_url":"https://api.github.com/repos/Parsely/pykafka/issues/538/labels{/name}","comments_url":"https://api.github.com/repos/Parsely/pykafka/issues/538/comments","events_url":"https://api.github.com/repos/Parsely/pykafka/issues/538/events","html_url":"https://github.com/Parsely/pykafka/issues/538","id":152752311,"number":538,"title":"Error discovering offset manager","user":{"login":"dvenza","id":5452054,"avatar_url":"https://avatars.githubusercontent.com/u/5452054?v=3","gravatar_id":"","url":"https://api.github.com/users/dvenza","html_url":"https://github.com/dvenza","followers_url":"https://api.github.com/users/dvenza/followers","following_url":"https://api.github.com/users/dvenza/following{/other_user}","gists_url":"https://api.github.com/users/dvenza/gists{/gist_id}","starred_url":"https://api.github.com/users/dvenza/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dvenza/subscriptions","organizations_url":"https://api.github.com/users/dvenza/orgs","repos_url":"https://api.github.com/users/dvenza/repos","events_url":"https://api.github.com/users/dvenza/events{/privacy}","received_events_url":"https://api.github.com/users/dvenza/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/Parsely/pykafka/labels/hazy","name":"hazy","color":"fbca04"},{"url":"https://api.github.com/repos/Parsely/pykafka/labels/needs%20response","name":"needs response","color":"fef2c0"},{"url":"https://api.github.com/repos/Parsely/pykafka/labels/question","name":"question","color":"cc317c"}],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":6,"created_at":"2016-05-03T11:29:31Z","updated_at":"2016-07-04T06:35:08Z","closed_at":"2016-07-01T18:47:50Z","body":"Using Kafka 0.9.0.1 and pykafka from pip, on python3. The topic has 2 replicas and two partitions on a two broker cluster.\r\n\r\nI don't know if it is related to #405 or to some other problem.\r\nWhen I create a consumer with this code:\r\n```\r\n    consumer = topic.get_simple_consumer(consumer_group='pq_consumer'.encode('utf-8'))\r\n    cursor = pq_conn.cursor()\r\n    for message in consumer:\r\n        if message is not None:\r\n            do_stuff(message)\r\n        consumer.commit_offsets()\r\n```\r\n\r\nI get this error:\r\n```\r\n2016-05-03 11:13:28,330 INFO pykafka.cluster (MainThread): Attempting to discover offset manager for consumer group 'b'pq_consumer''\r\n2016-05-03 11:13:28,332 ERROR pykafka.cluster (MainThread): Error discovering offset manager.\r\n2016-05-03 11:13:28,333 DEBUG pykafka.cluster (MainThread): Retrying offset manager discovery\r\n2016-05-03 11:13:29,035 INFO pykafka.handlers (0: pykafka.RequestHandler.worker for 192.168.45.252:9092): RequestHandler worker: exiting cleanly\r\n2016-05-03 11:13:29,325 INFO pykafka.handlers (3: pykafka.RequestHandler.worker for 192.168.45.252:9092): RequestHandler worker: exiting cleanly\r\n2016-05-03 11:13:30,337 ERROR pykafka.cluster (MainThread): Error discovering offset manager.\r\n2016-05-03 11:13:30,338 DEBUG pykafka.cluster (MainThread): Retrying offset manager discovery\r\n2016-05-03 11:13:34,345 ERROR pykafka.cluster (MainThread): Error discovering offset manager.\r\nTraceback (most recent call last):\r\n  File \"/home/ubuntu/zoe-logger/pq-consumer.py\", line 21, in <module>\r\n    main()\r\n  File \"/home/ubuntu/zoe-logger/pq_consumer/entrypoint.py\", line 87, in main\r\n    main_loop(kafka_topic, pq_conn)\r\n  File \"/home/ubuntu/zoe-logger/pq_consumer/entrypoint.py\", line 32, in main_loop\r\n    consumer = topic.get_simple_consumer(consumer_group='pq_consumer'.encode('utf-8'), auto_commit_enable=False)\r\n  File \"/home/ubuntu/pykafka/pykafka/topic.py\", line 192, in get_simple_consumer\r\n    **kwargs)\r\n  File \"/home/ubuntu/pykafka/pykafka/simpleconsumer.py\", line 182, in __init__\r\n    self._discover_group_coordinator()\r\n  File \"/home/ubuntu/pykafka/pykafka/simpleconsumer.py\", line 312, in _discover_group_coordinator\r\n    self._group_coordinator = self._cluster.get_group_coordinator(self._consumer_group)\r\n  File \"/home/ubuntu/pykafka/pykafka/cluster.py\", line 391, in get_group_coordinator\r\n    res = future.get(GroupCoordinatorResponse)\r\n  File \"/home/ubuntu/pykafka/pykafka/handlers.py\", line 68, in get\r\n    return response_cls(self.response)\r\n  File \"/home/ubuntu/pykafka/pykafka/protocol.py\", line 929, in __init__\r\n    self.raise_error(error_code, response)\r\n  File \"/home/ubuntu/pykafka/pykafka/protocol.py\", line 122, in raise_error\r\n    clsname, response))\r\npykafka.exceptions.GroupCoordinatorNotAvailable: Response Type: \"GroupCoordinatorResponse\"\tResponse: (15, -1, b'', -1)\r\n```\r\nI reference the unicode issue because of that first line: `'b'pq_consumer''` is clearly wrong, but the problem could be somewhere else."},"comment":{"url":"https://api.github.com/repos/Parsely/pykafka/issues/comments/230215501","html_url":"https://github.com/Parsely/pykafka/issues/538#issuecomment-230215501","issue_url":"https://api.github.com/repos/Parsely/pykafka/issues/538","id":230215501,"user":{"login":"mlyashkov","id":19308603,"avatar_url":"https://avatars.githubusercontent.com/u/19308603?v=3","gravatar_id":"","url":"https://api.github.com/users/mlyashkov","html_url":"https://github.com/mlyashkov","followers_url":"https://api.github.com/users/mlyashkov/followers","following_url":"https://api.github.com/users/mlyashkov/following{/other_user}","gists_url":"https://api.github.com/users/mlyashkov/gists{/gist_id}","starred_url":"https://api.github.com/users/mlyashkov/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mlyashkov/subscriptions","organizations_url":"https://api.github.com/users/mlyashkov/orgs","repos_url":"https://api.github.com/users/mlyashkov/repos","events_url":"https://api.github.com/users/mlyashkov/events{/privacy}","received_events_url":"https://api.github.com/users/mlyashkov/received_events","type":"User","site_admin":false},"created_at":"2016-07-04T06:35:07Z","updated_at":"2016-07-04T06:35:07Z","body":"@emmett9001, I try this:\r\n` import pykafka`\r\n`import logging`\r\n`logging.basicConfig(level=logging.NOTSET)`\r\n`cl = pykafka.KafkaClient(hosts='kafka201:9092, kafka202:9092, kafka203:9092')`\r\n\r\n> DEBUG:pykafka.cluster:Updating cluster, attempt 1/3\r\n> DEBUG:pykafka.connection:Connecting to kafka201:9092\r\n> DEBUG:pykafka.connection:Successfully connected to kafka201:9092\r\n> WARNING:pykafka.broker:Leader not available for topic '__consumer_offsets' partition 0.\r\n> INFO:pykafka.handlers:RequestHandler.stop: about to flush requests queue\r\n> INFO:pykafka.cluster:Discovered 3 brokers\r\n> DEBUG:pykafka.cluster:Discovered broker id 392: kafka203:9092\r\n> DEBUG:pykafka.connection:Connecting to kafka203:9092\r\n> DEBUG:pykafka.connection:Successfully connected to kafka203:9092\r\n> DEBUG:pykafka.cluster:Discovered broker id 393: kafka201:9092\r\n> DEBUG:pykafka.connection:Connecting to kafka201:9092\r\n> DEBUG:pykafka.connection:Successfully connected to kafka201:9092\r\n> DEBUG:pykafka.cluster:Discovered broker id 394: kafka202:9092\r\n> DEBUG:pykafka.connection:Connecting to kafka202:9092\r\n> DEBUG:pykafka.connection:Successfully connected to kafka202:9092\r\n> INFO:pykafka.cluster:Discovered 13 topics\r\n> DEBUG:pykafka.cluster:Discovered topic 'tmp'\r\n> DEBUG:pykafka.cluster:Discovered topic 'audiences_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'second_party_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'mobile_app_description_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'dsp_delivery_seance_features_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'remarketing_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'identify_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'url_class_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'site_index_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'test_topic'\r\n> DEBUG:pykafka.cluster:Discovered topic 'visitor_action_queue'\r\n> DEBUG:pykafka.cluster:Discovered topic 'search_queries_queue'\r\n> INFO:pykafka.handlers:RequestHandler worker: exiting cleanly\r\n\r\n`topic = cl.topics['site_index_queue']`\r\n> DEBUG:pykafka.connection:Connecting to kafka201:9092\r\n> DEBUG:pykafka.connection:Successfully connected to kafka201:9092\r\n> INFO:pykafka.handlers:RequestHandler.stop: about to flush requests queue\r\n> INFO:pykafka.topic:Adding 3 partitions\r\n> DEBUG:pykafka.topic:Adding partition site_index_queue/0\r\n> DEBUG:pykafka.topic:Adding partition site_index_queue/1\r\n> DEBUG:pykafka.topic:Adding partition site_index_queue/2\r\n> INFO:pykafka.handlers:RequestHandler worker: exiting cleanly\r\n\r\n`balanced_consumer = topic.get_balanced_consumer(consumer_group='test_group', auto_commit_enable=True, zookeeper_connect='kafka201:2181,kafka202:2181,kafka203:2181')`\r\n\r\n> Level 5:kazoo.client:ZK loop started\r\n> Level 5:kazoo.client:Skipping state change\r\n> INFO:kazoo.client:Connecting to kafka201:2181\r\n> Level 5:kazoo.client:    Using session_id: None session_passwd: 00000000000000000000000000000000\r\n> DEBUG:kazoo.client:Sending request(xid=None): Connect(protocol_version=0, last_zxid_seen=0, time_out=6000, session_id=0, passwd='\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00\\x00', read_only=None)\r\n> Level 5:kazoo.client:Read response Connect(protocol_version=0, last_zxid_seen=0, time_out=6000, session_id=96147589277990315, passwd='48%\\xb6gX\\xbb\\x13\\xf8\\x80\\x84_\\x00\\xbf\\xcb>', read_only=False)\r\n> Level 5:kazoo.client:Session created, session_id: 96147589277990315 session_passwd: 343825b66758bb13f880845f00bfcb3e\r\n>     negotiated session timeout: 6000\r\n>     connect timeout: 2000\r\n>     read timeout: 4000.0\r\n> INFO:kazoo.client:Zookeeper connection established, state: CONNECTED\r\n> DEBUG:kazoo.client:Sending request(xid=1): Exists(path='/consumers/test_group/owners/site_index_queue', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=1, zxid=270594036847, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=1): ZnodeStat(czxid=270589726248, mzxid=270589726248, ctime=1467375307214, mtime=1467375307214, version=0, cversion=15, aversion=0, ephemeralOwner=0, dataLength=0, numChildren=3, pzxid=270589741377)\r\n> DEBUG:kazoo.client:Sending request(xid=2): Create(path='/consumers/test_group/ids/mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363', data='site_index_queue', acl=[ACL(perms=31, acl_list=['ALL'], id=Id(scheme='world', id='anyone'))], flags=1)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=2, zxid=270594036848, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=2): u'/consumers/test_group/ids/mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363'\r\n> DEBUG:kazoo.client:Sending request(xid=3): GetChildren(path='/brokers/ids', watcher=<bound method ChildrenWatch._watcher of <kazoo.recipe.watchers.ChildrenWatch object at 0x7fee4a9a2710>>)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=3, zxid=270594036851, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=3): [u'392', u'393', u'394']\r\n> DEBUG:kazoo.client:Sending request(xid=4): GetChildren(path='/brokers/topics', watcher=<bound method ChildrenWatch._watcher of <kazoo.recipe.watchers.ChildrenWatch object at 0x7fee4a9a2890>>)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=4, zxid=270594036852, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=4): [u'url_class_queue', u'remarketing_queue', u'dsp_delivery_seance_features_queue', u'search_queries_queue', u'site_index_queue', u'second_party_queue', u'identify_queue', u'mobile_app_description_queue', u'test_topic', u'tmp', u'visitor_action_queue', u'audiences_queue', u'__consumer_offsets']\r\n> DEBUG:kazoo.client:Sending request(xid=5): GetChildren(path='/consumers/test_group/ids', watcher=<bound method ChildrenWatch._watcher of <kazoo.recipe.watchers.ChildrenWatch object at 0x7fee4a9a2910>>)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=5, zxid=270594036855, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=5): [u'mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363', u'mlyashkov-comp:23b510cf-698d-4da8-8509-09471e1811aa']\r\n> INFO:pykafka.balancedconsumer:Rebalancing consumer \"mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363\" for topic \"site_index_queue\".\r\n> DEBUG:kazoo.client:Sending request(xid=6): GetChildren(path='/consumers/test_group/ids', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=6, zxid=270594036856, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=6): [u'mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363', u'mlyashkov-comp:23b510cf-698d-4da8-8509-09471e1811aa']\r\n> DEBUG:kazoo.client:Sending request(xid=7): GetData(path=u'/consumers/test_group/ids/mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=7, zxid=270594036857, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=7): ('site_index_queue', ZnodeStat(czxid=270594036848, mzxid=270594036848, ctime=1467613336154, mtime=1467613336154, version=0, cversion=0, aversion=0, ephemeralOwner=96147589277990315, dataLength=16, numChildren=0, pzxid=270594036848))\r\n> DEBUG:kazoo.client:Sending request(xid=8): GetData(path=u'/consumers/test_group/ids/mlyashkov-comp:23b510cf-698d-4da8-8509-09471e1811aa', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=8, zxid=270594036860, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=8): ('site_index_queue', ZnodeStat(czxid=270589744395, mzxid=270589744395, ctime=1467376279362, mtime=1467376279362, version=0, cversion=0, aversion=0, ephemeralOwner=0, dataLength=16, numChildren=0, pzxid=270589744395))\r\n> INFO:pykafka.balancedconsumer:mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363: Balancing 2 participants for 3 partitions. Owning 1 partitions.\r\n> DEBUG:pykafka.balancedconsumer:My partitions: ['site_index_queue-394-1']\r\n> DEBUG:kazoo.client:Sending request(xid=9): GetChildren(path='/consumers/test_group/owners/site_index_queue', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=9, zxid=270594036865, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=9): [u'392-2', u'393-0', u'394-1']\r\n> DEBUG:kazoo.client:Sending request(xid=10): GetData(path='/consumers/test_group/owners/site_index_queue/392-2', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=10, zxid=270594036865, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=10): ('mlyashkov-comp:2dd374db-71eb-40cf-99a9-dd6502da8d7b', ZnodeStat(czxid=270589740386, mzxid=270589740386, ctime=1467376066940, mtime=1467376066940, version=0, cversion=0, aversion=0, ephemeralOwner=0, dataLength=51, numChildren=0, pzxid=270589740386))\r\n> DEBUG:kazoo.client:Sending request(xid=11): GetData(path='/consumers/test_group/owners/site_index_queue/393-0', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=11, zxid=270594036869, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=11): ('mlyashkov-comp:2dd374db-71eb-40cf-99a9-dd6502da8d7b', ZnodeStat(czxid=270589739687, mzxid=270589739687, ctime=1467376018941, mtime=1467376018941, version=0, cversion=0, aversion=0, ephemeralOwner=0, dataLength=51, numChildren=0, pzxid=270589739687))\r\n> DEBUG:kazoo.client:Sending request(xid=12): GetData(path='/consumers/test_group/owners/site_index_queue/394-1', watcher=None)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=12, zxid=270594036869, err=0)\r\n> DEBUG:kazoo.client:Received response(xid=12): ('mlyashkov-comp:2dd374db-71eb-40cf-99a9-dd6502da8d7b', ZnodeStat(czxid=270589741377, mzxid=270589741377, ctime=1467376119345, mtime=1467376119345, version=0, cversion=0, aversion=0, ephemeralOwner=0, dataLength=51, numChildren=0, pzxid=270589741377))\r\n> DEBUG:kazoo.client:Sending request(xid=13): Create(path='/consumers/test_group/owners/site_index_queue/394-1', data='mlyashkov-comp:f1429529-1b80-41d3-b94a-e73beaddd363', acl=[ACL(perms=31, acl_list=['ALL'], id=Id(scheme='world', id='anyone'))], flags=1)\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=13, zxid=270594036872, err=-110)\r\n> DEBUG:kazoo.client:Received error(xid=13) NodeExistsError((), {})\r\n> INFO:pykafka.cluster:Attempting to discover offset manager for consumer group 'test_group'\r\n> ERROR:pykafka.cluster:Error discovering offset manager.\r\n> DEBUG:pykafka.cluster:Retrying offset manager discovery\r\n> Level 5:kazoo.client:Sending request(xid=-2): Ping()\r\n> Level 5:kazoo.client:Received Ping\r\n> ERROR:pykafka.cluster:Error discovering offset manager.\r\n> DEBUG:pykafka.cluster:Retrying offset manager discovery\r\n> Level 5:kazoo.client:Sending request(xid=-2): Ping()\r\n> Level 5:kazoo.client:Received Ping\r\n> Level 5:kazoo.client:Sending request(xid=-2): Ping()\r\n> Level 5:kazoo.client:Received Ping\r\n> ERROR:pykafka.cluster:Error discovering offset manager.\r\n> ERROR:pykafka.balancedconsumer:Stopping consumer in response to error\r\n> Traceback (most recent call last):\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 306, in start\r\n>     self._rebalance()\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 603, in _rebalance\r\n>     self._update_member_assignment()\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 578, in _update_member_assignment\r\n>     if self._setup_internal_consumer(new_partitions):\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 359, in _setup_internal_consumer\r\n>     cns = self._get_internal_consumer(partitions=list(partitions), start=start)\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/balancedconsumer.py\", line 417, in _get_internal_consumer\r\n>     consumer_id=self._consumer_id\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 187, in __init__\r\n>     self._discover_group_coordinator()\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/simpleconsumer.py\", line 315, in _discover_group_coordinator\r\n>     self._group_coordinator = self._cluster.get_group_coordinator(self._consumer_group)\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/cluster.py\", line 390, in get_group_coordinator\r\n>     res = future.get(GroupCoordinatorResponse)\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/handlers.py\", line 68, in get\r\n>     return response_cls(self.response)\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/protocol.py\", line 929, in __init__\r\n>     self.raise_error(error_code, response)\r\n>   File \"/home/mlyashkov/anaconda2/lib/python2.7/site-packages/pykafka/protocol.py\", line 122, in raise_error\r\n>     clsname, response))\r\n> GroupCoordinatorNotAvailable: Response Type: \"GroupCoordinatorResponse\"\tResponse: (15, -1, '', -1)\r\n> DEBUG:pykafka.balancedconsumer:Stopping <pykafka.balancedconsumer.BalancedConsumer at 0x7fee4aa0d1d0 (consumer_group=test_group)>\r\n> DEBUG:kazoo.client:Sending request(xid=14): Close()\r\n> DEBUG:kazoo.client:Received EVENT: Watch(type=4, state=3, path=u'/consumers/test_group/ids')\r\n> Level 5:kazoo.client:Reading for header ReplyHeader(xid=14, zxid=270594037147, err=0)\r\n> Level 5:kazoo.client:Read close response\r\n> INFO:kazoo.client:Closing connection to kafka201:2181\r\n> INFO:kazoo.client:Zookeeper session lost, state: CLOSED\r\n> Level 5:kazoo.client:Connection stopped\r\n> DEBUG:pykafka.simpleconsumer:Finalising <pykafka.simpleconsumer.SimpleConsumer at 0x7fee4a9a2b10 (consumer_group=test_group)>\r\n\r\nMaybe I have a mistake in configuring clusters. I could not find the description of this error in Google. I hope for your help. \r\n"}}
{"actor":{"display_login":"mlyashkov"}}