Producer drops messages under heavy workload with batch_send=True

[yongkun]

The SimpleProducer dropped about 2/3 messages in my test when using producer sent to broker on another machine with Kafka 0.8.1.
I set batch size to 100, each message in my test is about 800 bytes.
buffer.size is default (100KB).

Test code:

from kafka.client import KafkaClient
from kafka.producer import SimpleProducer

kafka = KafkaClient("host:9092")

producer = SimpleProducer(kafka,
            async=False,
            #req_acks=Producer.ACK_AFTER_LOCAL_WRITE,
            #ack_timeout=Producer.DEFAULT_ACK_TIMEOUT, # 1000ms
            batch_send=True,
            batch_send_every_n=100,
            batch_send_every_t=1000)

data_file = "mylog.log"

count = 0
with open(data_file) as f:
    for line in f:
        producer.send_messages("my-topic", str(count) + line)
        count += 1

There is no problem when the batch_send is disabled.
I also tested the Java client, no messages dropped.

[dpkp]

i haven't tested, but hunch is that the main body of your test script is ending before the async thread is done. You might try at the very end something like calling producer.stop(timeout=None) or some timeout that seems reasonable given the data.

[wizzat]

I remember seeing behavior like this in some of the tests if you don't stop the async producer. It might be worth putting an atexit hook in to stop it for us.

-Mark

On May 12, 2014, at 22:35, Dana Powers notifications@github.com wrote:

i haven't tested, but hunch is that the main body of your test script is ending before the async thread is done. You might try at the very end something like calling producer.stop(timeout=None) or some timeout that seems reasonable given the data.

—
Reply to this email directly or view it on GitHub.

[yongkun]

It works well after adding producer.stop(timeout=None), thanks @dpkp . But I feel that it's confusing since I have already specified async=False. A hook may be better, as proposed by @wizzat .

[dpkp]

sure, though you are using async if you specify batch_send=True. There is no way to do batch synchronously :)

https://github.com/mumrah/kafka-python/blob/master/kafka/producer.py#L109-L110

[wizzat]

Depending on how you think about batching, it's possible to manually batch messages because producer.send_messages takes many messages and creates a single message set and dispatches it. I've found you have to be careful about max message size limits though.

[dpkp]

producers now register atexit handlers - #360