consume(timeout=1) takes 5 seconds

[elbaro]

Description

consume(num_messages=100, timeout=1) takes about 5 seconds.

How to reproduce

consumer.py

import time
from confluent_kafka import Consumer

consumer = Consumer({
    'bootstrap.servers': 'localhost',
    'group.id': 'example_consumer',
    'default.topic.config': {
        'auto.offset.reset': 'smallest'
     }
})

consumer.subscribe(['test_q'])

while True:
    while True:
        print('.')
        batch = consumer.consume(num_messages=100, timeout=1)
        if len(batch)>0:
            break
    print(len(batch))
    print('---')
    time.sleep(1)

producer.py

import time
from confluent_kafka import Producer

producer = Producer({
    'bootstrap.servers': 'localhost',
    'queue.buffering.max.messages': 10000000,  # 1e7
})

i = 0

while True:
    i += 1
    print(i)

    msg = 'msg ' + str(i)
    producer.poll(0)
    producer.produce('test_q', msg.encode(), callback=None)
    time.sleep(0.1)

open 3 terminals and run

1. python consumer.py
2. python producer.py
3. ./bin/kafka-console-consumer.sh --zookeeper localhost --topic test_q

You can see the immediate consumption in the terminal3.
The producer sends 10 messages/sec so I expect the consumer to return the batch of size 10.

However, terminal1 prints every 5 seconds. consumer.consume doesn't return for 5 seconds.

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()): ('0.11.4', 721920), ('0.11.4', 722175)
• Apache Kafka broker version: 2.12-1.1.0
• Client configuration: {...}
• Operating system:
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[rnpridgeon]

@elbaro, the test case you have provided is like comparing apples to oranges. If you want to compare performance between the java console consumer and the python consumer you should edit your code to look like the following:

from confluent_kafka import Consumer, KafkaError

consumer = Consumer({
    'bootstrap.servers': 'localhost',
    'group.id': 'example_consumer',
    'default.topic.config': {
        'auto.offset.reset': 'smallest'
     }
})

consumer.subscribe(['testq'])

while True:
    msg = consumer.poll(1.0)

    if msg is None:
        continue
    if msg.error():
        if msg.error().code() == KafkaError._PARTITION_EOF:
            continue
        else:
            print(msg.error())
            break

    print('{}'.format(msg.value().decode('utf-8')))

c.close()

The console consumer should also be started with --bootstrap-server localhost:9092 as opposed to --zookeeper. If you fail to use --bootstrap-server the console consumer will use the old consumer rather than the new consumer.

[elbaro]

@rnpridgeon Is consumer.poll(1.0) supposed to be faster than consumer.consume(num_messages=1, timeout=1.0)? It seems consume is just a batched version of poll.

If I use consume(num_messages=1, timeout=1.0), it is as fast as poll(1.0). I think the problem is the batching. If you set num_messages=100 and receives only 20 messages before the timeout, it should timeout and return the batch of 20 messages. However for some reason it does not timeout and keep waiting.

[rnpridgeon]

@elbaro, your results are skewed heavily because of the unnecessary sleeps you added into the code.

If you want to track the amount of time your consumer is waiting on a batch I recommend doing something like the following:

start = time.time()
 batch = consumer.consume(num_messages=100, timeout=1)
        if len(batch)>0:
            break
    print('{} in {} seconds'.format(len(msg), time.time()-start))

Get rid of the sleep on both the producer and the consumer, these are unnecessary and add artificial latency to your tests on both the produce and consume side. Once you have removed those you will find that batching works as expected.

[elbaro]

@rnpridgeon I can reproduce without time.sleep().

Consumer without time.sleep:

import time
from confluent_kafka import Consumer
consumer = Consumer({
    'bootstrap.servers': 'localhost',
    'group.id': 'example_consumer',
    'default.topic.config': {
        'auto.offset.reset': 'smallest'
    }
})
consumer.subscribe(['test_q'])
while True:
    while True:
        t0 = time.time()
        batch = consumer.consume(num_messages=100, timeo
ut=1)
        t = time.time()
        print('consume() took ', t-t0)
        if len(batch) > 0:
            break
    batch = [sample.value().decode() for sample in batch
 if (sample.error() is None)]
    print(len(batch), batch)
    #time.sleep(1)

Producer

import time
from confluent_kafka import Producer
producer = Producer({
    'bootstrap.servers': 'localhost',
    'queue.buffering.max.messages': 10000000,  # 1e7
})
i = 0
while True:
    i += 1
    print(i)
    msg = 'msg ' + str(i)
    producer.poll(0)
    producer.produce('test_q', msg.encode(), callback=No
ne)
    #time.sleep(0.1)
    if i==20:
        break

time.sleep(10) # wait for all msgs to be sent

Consumer Stdout

consume() took  1.0001533031463623
consume() took  1.0001201629638672
consume() took  1.0001556873321533
consume() took  1.0001211166381836
consume() took  1.000159502029419
consume() took  2.0007779598236084
20 ['msg 1', 'msg 2', 'msg 3', 'msg 4', 'msg 5', 'msg 6', 'msg 7', 'msg 8', 'msg 9', 'msg 10', 'msg 11', 'msg 12', 'msg 13', 'msg 14', 'msg 15', 'msg 16', 'msg 17', 'msg 18', 'msg 19', 'msg 20']
consume() took  1.0001561641693115
consume() took  1.0001680850982666
consume() took  1.9313828945159912
consume() took  1.0001683235168457
consume() took  1.0001590251922607

consume() should not take more than 1 second (timeout=1)

Also in my original examples if the consumer script finishes, the producer's consume() immediately returns. However if the producer keeps producing values, the consumer waits for the full batch (and wait for more than 1 second).

[rnpridgeon]

It would seem the timeout value provided to consume is actually how long to wait for a single message not how long to wait for a batch to be filled.

timeout (float) – Maximum time to block waiting for message, event or callback (default: infinite (-1)). (Seconds)

At a glance the following snippet appears to confirm this as well.
https://github.com/edenhill/librdkafka/blob/master/src/rdkafka_queue.c#L537-L551

This seems contrary however to the description in rdkafka.h
https://github.com/edenhill/librdkafka/blob/master/src/rdkafka.h#L2656-L2660

I'll follow up tomorrow after I've had time to look into this a bit more

[elbaro]

Ah, that may explain it.

[rnpridgeon]

I spoke with Magnus briefly about this and this does appear to be the case and its by design.

My previous comment is a bit misleading so allow me to clarify along with the reasoning for this behavior in librdkafka.

When I said "wait for a single message " what I really meant is a single message could reset the timeout if the previous fetch did not satisfy the num_messages limit.

Since messages are fetched and enqueued, in batches each iteration of the loop could satisfy your batching requirements. In the event the number of records fetched could not it will loop back around and try again for another full timeout. In other words timeout_ms is actually more of an idle timeout than a batching timeout. If the loop can't make any progress within the configured timeout threshold it will return leaving you with a partially filled buffer.

The documentation could be more clear on this in both librdkafka and python. I'll make a note of it to get that clarified.

[elbaro]

That's how I understood it.

If 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.

In 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)

[elbaro]

The upstream issue said it is a bug :p

[rnpridgeon]

Even better, I must have done a poor job explaining the situation to Magnus yesterday. Thanks for following up.