feat: retention example

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