produce.py

import random
import time
from confluent_kafka import Producer

options = {
    'bootstrap.servers': "localhost:29092",
    'retries': 10,
    'delivery.report.only.error': False,  # if set to True, will not be able be able to count successful delivery
    'message.max.bytes': 2097152
}
producer = Producer(**options)

count = 0

def on_delivered(err, msg):
    global count
    if err:
        print err, msg.value()
    else:
        print msg.offset()
        count += 1

with open('/usr/share/dict/words') as fh:
    for line in fh:
        word = line.strip()

        # produce is asynchronous
        producer.produce('words', word, on_delivery=on_delivered)
        #time.sleep(random.random())

    # need to force a flush so all data is sent
    producer.flush()

print "{} messages delivered".format(count)