feat: add python producer consumer example

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

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

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

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.