Manual Offset Commits & Exactly-Once Once Processing in Kafka Consumer using Python

In this video Manual Offset Commits & Exactly-Once Once Processing in Kafka Consumer is covered in detail using Python.

Note:
----------
When we spin up a consumer and the consumer has subscribed to a topic which already has some messages in diff partitions then the consumptions of messages will not be in order in newly spinned up consumer as the single consumer is consuming messages from more than one partition and message ordering not guaranteed across multiple partitions-- same phenomenon we observed in this demo also...

Also note , kafka is maintaining offset in partition level for a topic as different consumers are consuming messages from different partitions , so committed offsets for diff consumers will not be same , so it is not possible to have a single offset in topic level.
As a result when consumer rebalancing happens , then for any partition , the new consumer start consuming the messages where earlier consumer stopped & ensure no messages are missed.

Producer Code used:
----------------------------------------
from time import sleep
from json import dumps
from kafka import KafkaProducer

topic_name='hello_world1'
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],value_serializer=lambda x: dumps(x).encode('utf-8'))

for e in range(1000):
data = {'number' : e}
print(data)
producer.send(topic_name, value=data)
sleep(2)


Consumer Code:
-------------------------------
from kafka import KafkaConsumer
from kafka import TopicPartition , OffsetAndMetadata


import json


consumer = KafkaConsumer ('hello_world1',bootstrap_servers = ['localhost:9092'],
value_deserializer=lambda m: json.loads(m.decode('utf-8')),group_id='demo112215sgtrjwrykvjh',auto_offset_reset='earliest',
enable_auto_commit =False)


for message in consumer:
print(message)
print("The value is : {}".format(message.value))
print("The key is : {}".format(message.key))
print("The topic is : {}".format(message.topic))
print("The partition is : {}".format(message.partition))
print("The offset is : {}".format(message.offset))
print("The timestamp is : {}".format(message.timestamp))
tp=TopicPartition(message.topic,message.partition)
om = OffsetAndMetadata(message.offset+1, message.timestamp)
consumer.commit({tp:om})
print('*' * 100)



Prerequisite:
--------------------------
Consumer & Consumer Group in Kafka
https://youtu.be/QTi_oQhxARs
Kafka Consumer Groups CLI Demo | Kafka-Python
https://youtu.be/yKzdum_vX7k
Kafka Consumer using Python & Concept of Offset-Commit
https://youtu.be/da5_cH-Tx4c
In-depth intuition on Kafka Consumer Internal Architecture
https://youtu.be/UXxsLrt7c-g
In depth intuition on Manual Offset Commits & At Least Once Processing in Kafka Consumer
https://youtu.be/pavNd_LT9nI
In depth intuition on Manual Offset Commits & Exactly Once Processing in Kafka Consumer
https://youtu.be/IEmSzplZ-yo
In depth intuition on Manual Offset Commits & At-most Once Processing in Kafka Consumer
https://youtu.be/FwHc2iXsiPI

Check this playlist for more Data Engineering related videos:
https://youtube.com/playlist?list=PLjfRmoYoxpNopPjdACgS5XTfdjyBcuGku

Apache Kafka form scratch
https://youtube.com/playlist?list=PLjfRmoYoxpNrs0VmIq6mOTqXP52RfZdRf

Snowflake Complete Course from scratch with End-to-End Project with in-depth explanation--
https://doc.clickup.com/37466271/d/h/13qc4z-104/d4346819bd8d510

????????
YOU JUST NEED TO DO
3 THINGS to support my channel
LIKE
SHARE
&
SUBSCRIBE
TO MY YOUTUBE CHANNEL