From 15c93c71fb5a7356e21027a8ed6e2ec260e9834c Mon Sep 17 00:00:00 2001
From: ruinivist <ruinivist@proton.me>
Date: Sun, 25 Jan 2026 03:03:25 +0000
Subject: [PATCH] feat: retention example

---
 05-retention/consumer.py         | 27 +++++++++++++++++++++++++
 05-retention/docker-compose.yaml | 34 ++++++++++++++++++++++++++++++++
 05-retention/produce.py          | 30 ++++++++++++++++++++++++++++
 README.md                        | 14 +++++++++++++
 4 files changed, 105 insertions(+)
 create mode 100644 05-retention/consumer.py
 create mode 100644 05-retention/docker-compose.yaml
 create mode 100644 05-retention/produce.py

diff --git a/05-retention/consumer.py b/05-retention/consumer.py
new file mode 100644
index 0000000..e8c693c
--- /dev/null
+++ b/05-retention/consumer.py
@@ -0,0 +1,27 @@
+from kafka import KafkaConsumer
+
+topic_name = "ret-topic"
+group_id = "ret-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
+        text = message.value.decode('utf-8')
+        offset = message.offset
+        print(
+            f"Received: {text} from partition: {message.partition} at offset: {offset}")
+except KeyboardInterrupt:
+    print("Stopping consumer...")
+finally:
+    consumer.close()
diff --git a/05-retention/docker-compose.yaml b/05-retention/docker-compose.yaml
new file mode 100644
index 0000000..c5bab7f
--- /dev/null
+++ b/05-retention/docker-compose.yaml
@@ -0,0 +1,34 @@
+version: "3"
+
+services:
+    kafka:
+        image: apache/kafka:latest
+        container_name: kafka-ret
+        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/05-retention/produce.py b/05-retention/produce.py
new file mode 100644
index 0000000..02cfb50
--- /dev/null
+++ b/05-retention/produce.py
@@ -0,0 +1,30 @@
+# Generates traaffic to be put into the topic
+
+from kafka import KafkaProducer
+import time
+
+# make a topic
+topic_name = "ret-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
+        meta = 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} to partition {meta.partition} at offset {meta.offset}")
+        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 9e8da5b..a99c77f 100644
--- a/README.md
+++ b/README.md
@@ -71,3 +71,17 @@ DELETE topic:
 ```
 docker exec -it kafka-keys /opt/kafka/bin/kafka-topics.sh --delete --topic keys-topic --bootstrap-server localhost:9092
 ```
+
+# 05-retention
+
+5000 ms = 5 seconds of retention
+
+```
+sudo chown -R 1000:1000 kafka-data
+
+sudo docker exec kafka-ret /opt/kafka/bin/kafka-topics.sh --create --topic ret-topic --partitions 1 --bootstrap-server localhost:9092 --config retention.ms=5000 --config segment.ms=2000
+```
+
+Kafka moves on from one log file to another based on size or time. These are called "segments". Each segment is a file on disk. Nothing is EVER deleted from the active segment. Only when a segment is "closed" ( rolled over to a new segment ) can it be deleted based on retention policy.
+
+Just retention above won't work, I need to roll it over too.