how to close async producer

[darionyaphet]

Hi
I write a simple sender script reading a file and send to kafka 0.8.1 line by line when it is send to EOF but can't exit ! Actually this script is hang on ..

producer = SimpleProducer(client, async=True,
                          req_acks=SimpleProducer.ACK_AFTER_LOCAL_WRITE,
                          ack_timeout=2000,
                          batch_send=True)

for line in open(file_name,'r').readlines():
    producer.send_messages("topic.default.2",line)

producer.stop()
client.close()

Somebody could help me ? thanks a lot ~

[jswetzen]

I had a similar problem, in my case I couldn't make the script stop on KeyboardInterrupt.
In the end I just batched the messages manually by putting them in a list and sending them as keyword arguments. In your case it would be something like

lines = []

with open(file_name, 'r') as f:
    for line in f:
        lines.append(line)
        if len(lines) >= batch_size:
            producer.send_messages("topic", *lines)
            lines = []

It feels bad that I'm not using the batch functionality that's available, but I get twice the speed of using batch_send=True and the messages seem to be coming through all right.

[dpkp]

async producer code has been buggy, but we're hoping to clean it up for the next release. will update w/ results.

[dpkp]

fixes to async producer in #331 and #388 should make producer.stop() more reliable. You can also tune how long the producer will continue attempting to send queued messages after receiving the stop signal via async_stop_timeout