{"action":"created","issue":{"url":"https://api.github.com/repos/robinhood/faust/issues/225","repository_url":"https://api.github.com/repos/robinhood/faust","labels_url":"https://api.github.com/repos/robinhood/faust/issues/225/labels{/name}","comments_url":"https://api.github.com/repos/robinhood/faust/issues/225/comments","events_url":"https://api.github.com/repos/robinhood/faust/issues/225/events","html_url":"https://github.com/robinhood/faust/issues/225","id":384317402,"node_id":"MDU6SXNzdWUzODQzMTc0MDI=","number":225,"title":"Unable to fetch stream from kafka topic","user":{"login":"learnermaxRL","id":40789484,"node_id":"MDQ6VXNlcjQwNzg5NDg0","avatar_url":"https://avatars3.githubusercontent.com/u/40789484?v=4","gravatar_id":"","url":"https://api.github.com/users/learnermaxRL","html_url":"https://github.com/learnermaxRL","followers_url":"https://api.github.com/users/learnermaxRL/followers","following_url":"https://api.github.com/users/learnermaxRL/following{/other_user}","gists_url":"https://api.github.com/users/learnermaxRL/gists{/gist_id}","starred_url":"https://api.github.com/users/learnermaxRL/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/learnermaxRL/subscriptions","organizations_url":"https://api.github.com/users/learnermaxRL/orgs","repos_url":"https://api.github.com/users/learnermaxRL/repos","events_url":"https://api.github.com/users/learnermaxRL/events{/privacy}","received_events_url":"https://api.github.com/users/learnermaxRL/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":4,"created_at":"2018-11-26T12:53:41Z","updated_at":"2018-12-18T11:36:26Z","closed_at":"2018-11-27T05:37:48Z","author_association":"NONE","body":"## Checklist\r\n\r\n- [x] I have included information about relevant versions\r\n- [x] I have verified that the issue persists when using the `master` branch of Faust.\r\n\r\n## Steps to reproduce\r\n\r\n```\r\nimport faust\r\n\r\nclass Greeting(faust.Record):\r\n    # from_name: str\r\n    # to_name: str\r\n    time = str\r\n    image = str\r\n    camera = str\r\n    height = str\r\n    width = str\r\n\r\napp = faust.App('hello-app', broker='kafka://localhost:9092',topic_partitions=4)\r\ntopic = app.topic('camera_streams', value_type=Greeting)\r\n\r\n@app.agent(topic)\r\nasync def process(stream):\r\n    async for value in stream:\r\n         print(value)\r\n\r\n\r\nif __name__ == '__main__':\r\n    app.main()\r\n\r\n```\r\n Command - `faust -A stream_process worker -l info`\r\n\r\n## Expected behavior\r\n\r\nPrint values from kafka stream for the topic - \"camera_streams\",but instead faust shuts down with timeout error,my main goal is to consume the streams from kafka and process it using some function and then finally dump it in a sink.\r\n\r\n## Actual behavior\r\n\r\nTimeout,and is not able to connect and print the stream values from kafka topic\r\n\r\n## Full traceback\r\n\r\nPaste the full traceback (if there is any)\r\n\r\n```pytb\r\n\r\n┌ƒaµS† v1.3.2───────────────────────────────────────────────────────────────────────────────────────────┐\r\n│ id        │ hello-app                                                                                 │\r\n│ transport │ kafka://localhost:9092                                                                    │\r\n│ store     │ memory:                                                                                   │\r\n│ web       │ http://localhost:6066/                                                                    │\r\n│ log       │ -stderr- (info)                                                                           │\r\n│ pid       │ 14515                                                                                     │\r\n│ hostname  │ ms                                                                                     │\r\n│ platform  │ CPython 3.6.6 (Linux x86_64)                                                              │\r\n│ drivers   │ aiokafka=0.4.19 aiohttp=3.4.4                                                             │\r\n│ datadir   │ /home/ms/Downloads/faster_rcnn_inception_v2_coco_2018_01_28/utils_ob/hello-app-data    │\r\n│ appdir    │ /home/ms/Downloads/faster_rcnn_inception_v2_coco_2018_01_28/utils_ob/hello-app-data/v1 │\r\n└───────────┴───────────────────────────────────────────────────────────────────────────────────────────┘\r\n[2018-11-26 18:14:54,510: INFO]: [^Worker]: Starting...\r\n[2018-11-26 18:14:54,514: INFO]: [^-App]: Starting...\r\n[2018-11-26 18:14:54,514: INFO]: [^--Monitor]: Starting...\r\n[2018-11-26 18:14:54,514: INFO]: [^CacheBackend]: Starting...\r\n[2018-11-26 18:14:54,514: INFO]: [^Web]: Starting...\r\n[2018-11-26 18:14:54,518: INFO]: [^-ServerThread]: Starting...\r\n[2018-11-26 18:14:54,519: INFO]: [^Web]: Serving on http://localhost:6066/\r\n[2018-11-26 18:14:54,520: INFO]: [^--Producer]: Starting...\r\n[2018-11-26 18:14:54,633: INFO]: [^--Consumer]: Starting...\r\n[2018-11-26 18:14:54,749: INFO]: [^--LeaderAssignor]: Starting...\r\n[2018-11-26 18:14:54,750: INFO]: [^--Producer]: Creating topic hello-app-__assignor-__leader\r\n[2018-11-26 18:15:34,758: ERROR]: [^Worker]: Error: RequestTimedOutError()\r\nTraceback (most recent call last):\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/aiokafka/client.py\", line 432, in send\r\n    result = yield from future\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/asyncio/tasks.py\", line 362, in wait_for\r\n    raise futures.TimeoutError()\r\nconcurrent.futures._base.TimeoutError\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nTraceback (most recent call last):\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/worker.py\", line 191, in execute_from_commandline\r\n    self.loop.run_until_complete(self.start())\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/asyncio/base_events.py\", line 468, in run_until_complete\r\n    return future.result()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/worker.py\", line 243, in start\r\n    await super().start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 706, in start\r\n    await self._default_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 713, in _default_start\r\n    await self._actually_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 729, in _actually_start\r\n    await child.maybe_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 752, in maybe_start\r\n    await self.start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 706, in start\r\n    await self._default_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 713, in _default_start\r\n    await self._actually_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 729, in _actually_start\r\n    await child.maybe_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 752, in maybe_start\r\n    await self.start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 706, in start\r\n    await self._default_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 713, in _default_start\r\n    await self._actually_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 724, in _actually_start\r\n    await self.on_start()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/faust/assignor/leader_assignor.py\", line 20, in on_start\r\n    await leader_topic.maybe_declare()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/utils/futures.py\", line 36, in __call__\r\n    result = await self.fun(*self.args, **self.kwargs)\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/faust/topics.py\", line 388, in maybe_declare\r\n    await self.declare()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/faust/topics.py\", line 406, in declare\r\n    retention=self.retention,\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/faust/transport/drivers/aiokafka.py\", line 610, in create_topic\r\n    ensure_created=ensure_created,\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/faust/transport/drivers/aiokafka.py\", line 704, in _create_topic\r\n    await wrap()\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/utils/futures.py\", line 36, in __call__\r\n    result = await self.fun(*self.args, **self.kwargs)\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/faust/transport/drivers/aiokafka.py\", line 754, in _really_create_topic\r\n    timeout=timeout)\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/faust/transport/drivers/aiokafka.py\", line 720, in _get_controller_node\r\n    timeout=timeout,\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 618, in wait\r\n    return await self._wait_one(coros[0], timeout=timeout)\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 684, in _wait_one\r\n    results = await self.wait_first(coro, timeout=timeout)\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/mode/services.py\", line 662, in wait_first\r\n    f.result()  # propagate exceptions\r\n  File \"/home/ms/anaconda3/envs/tf_env/lib/python3.6/site-packages/aiokafka/client.py\", line 439, in send\r\n    raise RequestTimedOutError()\r\nrhkafka.errors.RequestTimedOutError: [Error 7] RequestTimedOutError\r\n[2018-11-26 18:15:34,768: INFO]: [^Worker]: Stopping...\r\n[2018-11-26 18:15:34,769: INFO]: [^-App]: Stopping...\r\n[2018-11-26 18:15:34,769: INFO]: [^-App]: Flush producer buffer...\r\n[2018-11-26 18:15:34,769: INFO]: [^--TableManager]: Stopping...\r\n[2018-11-26 18:15:34,770: INFO]: [^--Fetcher]: Stopping...\r\n[2018-11-26 18:15:34,770: INFO]: [^--Fetcher]: -Stopped!\r\n[2018-11-26 18:15:34,771: INFO]: [^--TableManager]: -Stopped!\r\n[2018-11-26 18:15:34,771: INFO]: [^--Conductor]: Stopping...\r\n[2018-11-26 18:15:34,771: INFO]: [^--Conductor]: -Stopped!\r\n[2018-11-26 18:15:34,772: INFO]: [^AgentManager]: Stopping...\r\n[2018-11-26 18:15:34,772: INFO]: [^Agent: stream_process.process]: Stopping...\r\n[2018-11-26 18:15:34,773: INFO]: [^Agent: stream_process.process]: -Stopped!\r\n[2018-11-26 18:15:34,773: INFO]: [^AgentManager]: -Stopped!\r\n[2018-11-26 18:15:34,773: INFO]: [^--ReplyConsumer]: Stopping...\r\n[2018-11-26 18:15:34,774: INFO]: [^--ReplyConsumer]: -Stopped!\r\n[2018-11-26 18:15:34,774: INFO]: [^--LeaderAssignor]: Stopping...\r\n[2018-11-26 18:15:34,774: INFO]: [^--LeaderAssignor]: -Stopped!\r\n[2018-11-26 18:15:34,775: INFO]: [^--Consumer]: Stopping...\r\n[2018-11-26 18:15:34,779: INFO]: [^--Consumer]: -Stopped!\r\n[2018-11-26 18:15:34,780: INFO]: [^--Producer]: Stopping...\r\n[2018-11-26 18:15:34,781: INFO]: [^--Producer]: -Stopped!\r\n[2018-11-26 18:15:34,782: INFO]: [^Web]: Stopping...\r\n[2018-11-26 18:15:34,782: INFO]: [^-ServerThread]: Stopping...\r\n[2018-11-26 18:15:34,783: INFO]: [^Web]: Closing server\r\n[2018-11-26 18:15:34,784: INFO]: [^Web]: Waiting for server to close handle\r\n[2018-11-26 18:15:34,784: INFO]: [^Web]: Shutting down web application\r\n[2018-11-26 18:15:34,784: INFO]: [^Web]: Waiting for handler to shut down\r\n[2018-11-26 18:15:34,785: INFO]: [^Web]: Cleanup\r\n[2018-11-26 18:15:34,785: INFO]: [^-ServerThread]: -Stopped!\r\n[2018-11-26 18:15:34,786: INFO]: [^Web]: -Stopped!\r\n[2018-11-26 18:15:34,786: INFO]: [^CacheBackend]: Stopping...\r\n[2018-11-26 18:15:34,786: INFO]: [^CacheBackend]: -Stopped!\r\n[2018-11-26 18:15:34,787: INFO]: [^--Monitor]: Stopping...\r\n[2018-11-26 18:15:34,788: INFO]: [^--Monitor]: -Stopped!\r\n[2018-11-26 18:15:34,788: INFO]: [^-App]: -Stopped!\r\n[2018-11-26 18:15:34,789: INFO]: [^Worker]: -Stopped!\r\n[2018-11-26 18:15:34,789: INFO]: [^Worker]: Gathering service tasks...\r\n[2018-11-26 18:15:34,790: INFO]: [^Worker]: Gathering all futures...\r\n[2018-11-26 18:15:35,892: INFO]: [^Worker]: Closing event loop\r\n\r\n```\r\n# Versions\r\n\r\n* Python version - 3.6\r\n* Faust version - 1.3.2\r\n* Operating system - ubuntu 16.04\r\n* Kafka version - kafka_2.11-0.9.0.0\r\n* RocksDB version (if applicable)"},"comment":{"url":"https://api.github.com/repos/robinhood/faust/issues/comments/448191322","html_url":"https://github.com/robinhood/faust/issues/225#issuecomment-448191322","issue_url":"https://api.github.com/repos/robinhood/faust/issues/225","id":448191322,"node_id":"MDEyOklzc3VlQ29tbWVudDQ0ODE5MTMyMg==","user":{"login":"learnermaxRL","id":40789484,"node_id":"MDQ6VXNlcjQwNzg5NDg0","avatar_url":"https://avatars3.githubusercontent.com/u/40789484?v=4","gravatar_id":"","url":"https://api.github.com/users/learnermaxRL","html_url":"https://github.com/learnermaxRL","followers_url":"https://api.github.com/users/learnermaxRL/followers","following_url":"https://api.github.com/users/learnermaxRL/following{/other_user}","gists_url":"https://api.github.com/users/learnermaxRL/gists{/gist_id}","starred_url":"https://api.github.com/users/learnermaxRL/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/learnermaxRL/subscriptions","organizations_url":"https://api.github.com/users/learnermaxRL/orgs","repos_url":"https://api.github.com/users/learnermaxRL/repos","events_url":"https://api.github.com/users/learnermaxRL/events{/privacy}","received_events_url":"https://api.github.com/users/learnermaxRL/received_events","type":"User","site_admin":false},"created_at":"2018-12-18T11:36:26Z","updated_at":"2018-12-18T11:36:26Z","author_association":"NONE","body":"Aplogies for the wrong version. Please use kafka_2.11-2.1.0 binary version.It should work"}}
{"actor":{"display_login":"learnermaxRL"}}