producer.py

# confluent kafka(.8.2) producer
import json
from confluent_kafka import Producer



class KafkaProducer:
    def __init__(self, hosts):
        self.hosts = hosts
        self.producer = Producer(
            {'bootstrap.servers': hosts, 'api.version.request': False, 'broker.version.fallback': '0.8.2'}
        )

    def send(self, topic, message):
        print("sending to hosts" + self.hosts)
        print("Sending message..... " + str(message))
        self.producer.produce(topic, json.dumps(message).encode('utf-8'))
        self.producer.flush()