From f8d7806ae17bc47d75236409238aa26320758b19 Mon Sep 17 00:00:00 2001
From: ruinivist <ruinivist@proton.me>
Date: Sun, 25 Jan 2026 01:27:46 +0000
Subject: [PATCH] feat: add python producer consumer example

---
 01-broken-setup/README.md          |  0
 02-python-conn/consumer.py         | 25 ++++++++++++++++++++++
 02-python-conn/docker-compose.yaml | 34 ++++++++++++++++++++++++++++++
 02-python-conn/produce.py          | 29 +++++++++++++++++++++++++
 README.md                          |  2 ++
 5 files changed, 90 insertions(+)
 delete mode 100644 01-broken-setup/README.md
 create mode 100644 02-python-conn/consumer.py
 create mode 100644 02-python-conn/docker-compose.yaml
 create mode 100644 02-python-conn/produce.py

diff --git a/01-broken-setup/README.md b/01-broken-setup/README.md
deleted file mode 100644
index e69de29..0000000
diff --git a/02-python-conn/consumer.py b/02-python-conn/consumer.py
new file mode 100644
index 0000000..e93360a
--- /dev/null
+++ b/02-python-conn/consumer.py
@@ -0,0 +1,25 @@
+from kafka import KafkaConsumer
+
+topic_name = "test-topic"
+group_id = "test-group"
+
+consumer = KafkaConsumer(
+    topic_name,
+    bootstrap_servers='localhost:9092',  # advertised listener
+    auto_offset_reset='earliest',  # start from the beginning if no offset is committed
+    enable_auto_commit=True,  # read = commit, this is True by default
+    # who am I in the consumer group, without it you'll always read from beginning
+    # as kafka does not know if you have read before as you had not name for yourself
+    group_id=group_id
+
+)
+
+try:
+    for message in consumer:
+        # message is of type ConsumerRecord and data is in value which is bytes
+        message = message.value.decode('utf-8')
+        print(f"Received: {message}")
+except KeyboardInterrupt:
+    print("Stopping consumer...")
+finally:
+    consumer.close()
diff --git a/02-python-conn/docker-compose.yaml b/02-python-conn/docker-compose.yaml
new file mode 100644
index 0000000..ad1a263
--- /dev/null
+++ b/02-python-conn/docker-compose.yaml
@@ -0,0 +1,34 @@
+version: "3"
+
+services:
+    kafka:
+        image: apache/kafka:latest
+        container_name: kafka-python-conn
+        ports:
+            - "127.0.0.1:9092:9092" # localhost interface,host:container
+        environment:
+            # who am I in the cluster
+            KAFKA_NODE_ID: 1 # node identifier in cluster
+            KAFKA_PROCESS_ROLES: broker,controller # Combined broker + controller (KRaft mode, no Zookeeper)
+            # how to vote in the cluster, service name at line 4
+            KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 # Controller nodes for quorum (node_id@host:port)
+
+            # Listener configuration
+            # what interfaces and ports on the container layer ( not host )
+            # PLAINTEXT and CONTROLLER are listener names, can be anything as long as
+            # referenced the correct ones after
+            KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 # Ports Kafka listens on internally
+            # outside container layer ( host )
+            KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 # Address advertised to clients
+            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT # encryption settings for listeners
+            KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER # listener name for controller communication
+
+            KAFKA_LOG_DIRS: /var/lib/kafka/data # Directory for message data storage
+
+            # when running locally you NEED this else consumer can't read via group
+            # as by default it wants 3 replicas for these internal topics and errors out
+            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 # replication factor for offsets topic
+            KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 # replication factor for transaction log
+            KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 # how many replicas must acknowledge a write
+        volumes:
+            - ./kafka-data:/var/lib/kafka/data # Persist data across container restarts
diff --git a/02-python-conn/produce.py b/02-python-conn/produce.py
new file mode 100644
index 0000000..c006c78
--- /dev/null
+++ b/02-python-conn/produce.py
@@ -0,0 +1,29 @@
+# Generates traaffic to be put into the topic
+
+from kafka import KafkaProducer
+import time
+
+# make a topic
+topic_name = "test-topic"
+
+# 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()
diff --git a/README.md b/README.md
index 66e8543..491cab6 100644
--- a/README.md
+++ b/README.md
@@ -14,3 +14,5 @@ version=1
 ```
 
 If this file is missing, the docker image will make one. But in case you run multiple brokers, all need to have the same cluster.id. So they can talk to each other. This is known as generate/format Kraft storage.
+
+Neede to run: `sudo chown -R 1000:1000 kafka-data/` to give the kafka user ( uid 1000 ) permission to write to the volume mounted directory. First 1000 is user id, second 1000 is group id.