# Generates traaffic to be put into the topic

from kafka import KafkaProducer
from kafka.errors import KafkaError
import time

# make a topic
topic_name = "rep-topic"  # same as the one made via script

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092', 'localhost:9094', 'localhost:9096'],
    acks='all',  # can be 0 (no ack), 1 (leader ack), all (all replicas ack)
    # doing all makes it use the compose setting of min.insync.replicas
    retries=5  # number of retries if produce fails
    # even if you set in sync replicas in compose, you still need to set acks all here
)

try:
    i = 0
    while True:
        message = f"Message {i}"
        meta = producer.send(topic_name, value=message.encode(
            'utf-8')).get()  # wait for ack
        print(
            f"Sent: {message} to partition {meta.partition} at offset {meta.offset}")
        i += 1
        time.sleep(1)  # 1 message per second
except KafkaError as ke:
    # if two replicas down, not enough replicas error is thrown
    print(f"Kafka error occurred: {ke}")
except KeyboardInterrupt:
    print("Stopping producer...")
finally:
    producer.close()