Kafka_Demo_Python/Producer.py at main · hams71/Kafka_Demo_Python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

from kafka import KafkaProducer
from kafka.errors import KafkaError
import json
from Data_Generator import get_reg_user
import time

def json_serializer(data):
    return json.dumps(data).encode("utf-8")

# only one producer for now we can have multiple producer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                        key_serializer=json_serializer,
                        value_serializer=json_serializer,
                        )
# for this we need to start a new kafka server in a new
# terminal and provide a new server.properties file
# which will have diff broker.id, port, log folder

# we can also give this bootstrap above as a list

producer1 = KafkaProducer(bootstrap_servers=['localhost:9093'],
                        key_serializer=json_serializer,
                        value_serializer=json_serializer,
                        )


if __name__ == "__main__":
    while 1==1:
        try:
            reg_user = get_reg_user()
            print(reg_user)
            metadata = producer.send(topic="Reg_User", key="first", value=reg_user)

        except KafkaError:
            pass

        try:
            reg_user = get_reg_user()
            print(reg_user)
            metadata = producer.send(topic="Data_User", key="second", value=reg_user)

        except KafkaError:
            pass


        # sync call will block and get metadata
        # try:
        #     record_metadata = metadata.get(timeout=10)
        # except KafkaError:
        #     pass

        # # Successful result returns assigned partition and offset
        # # These are used for sync calls only will not work for async
        # print (record_metadata.topic)
        # print (record_metadata.partition)
        # print (record_metadata.offset)

        time.sleep(3)