producer.py

from flask import Flask, request
from confluent_kafka import Producer
import json
import logging

logging.basicConfig(format='%(process)d >> %(asctime)s - %(message)s', level=logging.INFO)

app = Flask(__name__)


def acknowledgement(err, msg):
    # function to handle acknowledgement from kafka server
    if err is not None:
        logging.info("Failed to deliver message: {0}: {1}"
                     .format(msg.value(), err.str()))
    else:
        logging.info("Message produced: {0}".format(msg.value()))


p = Producer({'bootstrap.servers': 'localhost:9092'})


@app.route("/data", methods=["POST"])
def json_example():
    req = request.get_json()
    print(req)
    p.produce(topic="car_data", key=req["id"], value=json.dumps(req), callback=acknowledgement)
    # Wait for all messages in the Producer queue to be delivered
    p.flush(30)
    return "OK", 200


if __name__ == "__main__":
    app.run(port=9000)