# 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()