How do you handle duplicate messages in Kafka?

To handle message duplicates in Kafka, I used several approaches:

1. Idempotency of the consumer — storing unique message identifiers (e.g., messageId) in a database or cache to ignore reprocessing upon receiving the same message again.

2. Using message keys and partitioning — this helps organize messages and reduces the likelihood of duplication.

3. Idempotent database writes — for example, using UPSERT operations or checking for existing records before inserting.

An example of an idempotent consumer in Python using Redis to store processed IDs:

import redis

r = redis.Redis()

def process_message(msg):
    message_id = msg['id']
    if r.sismember('processed_messages', message_id):
        return  # message already processed
    # process the message
    # ...
    r.sadd('processed_messages', message_id)

This approach helps avoid reprocessing the same messages.