30 lines
891 B
Python
30 lines
891 B
Python
# Generates traaffic to be put into the topic
|
|
|
|
from kafka import KafkaProducer
|
|
import time
|
|
|
|
# make a topic
|
|
topic_name = "rebal-topic" # same as the one made via script
|
|
|
|
# producer instance
|
|
producer = KafkaProducer(
|
|
bootstrap_servers='localhost:9092') # advertised listener
|
|
|
|
try:
|
|
i = 0
|
|
while True:
|
|
# i NEED to encode to bytes, when I did not, failed with this assert
|
|
# assert type(value_bytes) in (bytes, bytearray, memoryview, type(None))
|
|
message = f"Message {i}"
|
|
# or I can define a value_serializer
|
|
producer.send(topic_name, value=message.encode(
|
|
'utf-8')).get() # wait for ack
|
|
# on the producer to do the encoding automatically
|
|
print(f"Sent: {message}")
|
|
i += 1
|
|
time.sleep(1) # 1 message per second
|
|
except KeyboardInterrupt:
|
|
print("Stopping producer...")
|
|
finally:
|
|
producer.close()
|