{"action":"created","issue":{"url":"https://api.github.com/repos/confluentinc/confluent-kafka-python/issues/407","repository_url":"https://api.github.com/repos/confluentinc/confluent-kafka-python","labels_url":"https://api.github.com/repos/confluentinc/confluent-kafka-python/issues/407/labels{/name}","comments_url":"https://api.github.com/repos/confluentinc/confluent-kafka-python/issues/407/comments","events_url":"https://api.github.com/repos/confluentinc/confluent-kafka-python/issues/407/events","html_url":"https://github.com/confluentinc/confluent-kafka-python/issues/407","id":334574844,"node_id":"MDU6SXNzdWUzMzQ1NzQ4NDQ=","number":407,"title":"consume(timeout=1) takes 5 seconds","user":{"login":"elbaro","id":1851290,"node_id":"MDQ6VXNlcjE4NTEyOTA=","avatar_url":"https://avatars3.githubusercontent.com/u/1851290?v=4","gravatar_id":"","url":"https://api.github.com/users/elbaro","html_url":"https://github.com/elbaro","followers_url":"https://api.github.com/users/elbaro/followers","following_url":"https://api.github.com/users/elbaro/following{/other_user}","gists_url":"https://api.github.com/users/elbaro/gists{/gist_id}","starred_url":"https://api.github.com/users/elbaro/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/elbaro/subscriptions","organizations_url":"https://api.github.com/users/elbaro/orgs","repos_url":"https://api.github.com/users/elbaro/repos","events_url":"https://api.github.com/users/elbaro/events{/privacy}","received_events_url":"https://api.github.com/users/elbaro/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":7,"created_at":"2018-06-21T16:48:19Z","updated_at":"2018-06-27T11:47:58Z","closed_at":null,"author_association":"NONE","body":"Description\r\n===========\r\n`consume(num_messages=100, timeout=1)` takes about 5 seconds.\r\n\r\n\r\nHow to reproduce\r\n================\r\n\r\nconsumer.py\r\n```py\r\nimport time\r\nfrom confluent_kafka import Consumer\r\n\r\nconsumer = Consumer({\r\n    'bootstrap.servers': 'localhost',\r\n    'group.id': 'example_consumer',\r\n    'default.topic.config': {\r\n        'auto.offset.reset': 'smallest'\r\n     }\r\n})\r\n\r\nconsumer.subscribe(['test_q'])\r\n\r\nwhile True:\r\n    while True:\r\n        print('.')\r\n        batch = consumer.consume(num_messages=100, timeout=1)\r\n        if len(batch)>0:\r\n            break\r\n    print(len(batch))\r\n    print('---')\r\n    time.sleep(1)\r\n```\r\n\r\nproducer.py\r\n```py\r\nimport time\r\nfrom confluent_kafka import Producer\r\n\r\nproducer = Producer({\r\n    'bootstrap.servers': 'localhost',\r\n    'queue.buffering.max.messages': 10000000,  # 1e7\r\n})\r\n\r\ni = 0\r\n\r\nwhile True:\r\n    i += 1\r\n    print(i)\r\n\r\n    msg = 'msg ' + str(i)\r\n    producer.poll(0)\r\n    producer.produce('test_q', msg.encode(), callback=None)\r\n    time.sleep(0.1)\r\n```\r\n\r\nopen 3 terminals and run\r\n\r\n1. `python consumer.py`\r\n2. `python producer.py`\r\n3. `./bin/kafka-console-consumer.sh --zookeeper localhost --topic test_q`\r\n\r\nYou can see the immediate consumption in the terminal3.\r\nThe producer sends 10 messages/sec so I expect the consumer to return the batch of size 10.\r\n\r\nHowever, terminal1 prints every 5 seconds. consumer.consume doesn't return for 5 seconds.\r\n\r\n\r\n\r\nChecklist\r\n=========\r\nPlease provide the following information:\r\n\r\n - [x] confluent-kafka-python and librdkafka version (`confluent_kafka.version()` and `confluent_kafka.libversion()`): ('0.11.4', 721920), ('0.11.4', 722175)\r\n - [x] Apache Kafka broker version: 2.12-1.1.0\r\n - [ ] Client configuration: `{...}`\r\n - [ ] Operating system:\r\n - [ ] Provide client logs (with `'debug': '..'` as necessary)\r\n - [ ] Provide broker log excerpts\r\n - [ ] Critical issue\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/confluentinc/confluent-kafka-python/issues/comments/400643907","html_url":"https://github.com/confluentinc/confluent-kafka-python/issues/407#issuecomment-400643907","issue_url":"https://api.github.com/repos/confluentinc/confluent-kafka-python/issues/407","id":400643907,"node_id":"MDEyOklzc3VlQ29tbWVudDQwMDY0MzkwNw==","user":{"login":"elbaro","id":1851290,"node_id":"MDQ6VXNlcjE4NTEyOTA=","avatar_url":"https://avatars3.githubusercontent.com/u/1851290?v=4","gravatar_id":"","url":"https://api.github.com/users/elbaro","html_url":"https://github.com/elbaro","followers_url":"https://api.github.com/users/elbaro/followers","following_url":"https://api.github.com/users/elbaro/following{/other_user}","gists_url":"https://api.github.com/users/elbaro/gists{/gist_id}","starred_url":"https://api.github.com/users/elbaro/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/elbaro/subscriptions","organizations_url":"https://api.github.com/users/elbaro/orgs","repos_url":"https://api.github.com/users/elbaro/repos","events_url":"https://api.github.com/users/elbaro/events{/privacy}","received_events_url":"https://api.github.com/users/elbaro/received_events","type":"User","site_admin":false},"created_at":"2018-06-27T11:47:58Z","updated_at":"2018-06-27T11:47:58Z","author_association":"NONE","body":"That's how I understood it.\r\n\r\nIf my queue has a ton of unprocessed messages, this problem does not happen. The problem happens if you call `consume(batch=100, timeout=1.0)` and the producer sends a message every 0.9 second, then the consume call doesn't give up(?) and may take up to 100*0.9 seconds. (Is this right?) `time.sleep(0.1)` was to simulate the slow producer.\r\n\r\nIn my case, sometimes the consumer is faster than the provider, and vice versa. There may be a burst of messages, and processing one message take long time (>1s). So if there is an incomplete batch, it is preferred to process them first rather than wait for the full batch with idle CPUs. `batch_timeout_ms` would be much useful in this scenario. (but my case might be uncommon)\r\n"}}
{"actor":{"display_login":"elbaro"}}