Kafka with Python: a Python Backend guide for developers

Introduction

Event streaming is now foundational to most backend systems, but integrating Kafka into Python services can feel overwhelming if you've never worked with message brokers before. Kafka with Python: a Python Backend guide for developers walks you through the patterns and libraries that work in production, so your team can reliably produce and consume events from Python services without the common pitfalls that slow teams down.

We'll cover confluent-kafka for performance, keying strategies for ordering, offset management, and idempotent processing—the same techniques we use across trading systems, ride-hailing backends, and e-commerce platforms where message order and exactly-once semantics matter.

Why confluent-kafka Over kafka-python

The Python ecosystem offers a few Kafka clients, but confluent-kafka is the production choice for most teams. It's a thin wrapper around librdkafka, a C library maintained by Confluent, which means it's fast, battle-tested, and used by companies handling billions of messages daily.

The pure Python alternative, kafka-python, is simpler to reason about but significantly slower and less reliable under high throughput. When you're processing thousands of messages per second—as you will in ride-hailing or trading—that difference becomes critical. Confluent's client also handles offset management, rebalancing, and network failures more gracefully out of the box.

Install it via pip:

pip install confluent-kafka

Here's a minimal producer setup:

from confluent_kafka import Producer
import json

producer = Producer({
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'python-backend-producer'
})

def delivery_report(err, msg):
    if err is not None:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

event = {'order_id': 12345, 'status': 'placed', 'timestamp': 1234567890}
producer.produce(
    topic='orders',
    value=json.dumps(event).encode('utf-8'),
    callback=delivery_report
)
producer.flush()  # Wait for all messages to be sent

The callback fires after Kafka acknowledges the message. This is where you log failures or trigger alerting. Always flush before shutdown to avoid losing buffered messages.

Use Keys for Ordering Per Entity

One of the biggest mistakes we see is treating Kafka as a simple queue. It's not. Kafka guarantees order only within a single partition, and all messages with the same key go to the same partition. If you don't set a key, messages are round-robin distributed, and you lose ordering guarantees entirely.

In a ride-hailing system, imagine events for a single ride: ride_requested, driver_assigned, ride_started, ride_completed. If those events arrive out of order to a consumer, your ride state machine breaks. The solution: use the ride ID as the key.

from confluent_kafka import Producer
import json

producer = Producer({'bootstrap.servers': 'localhost:9092'})

def produce_ride_event(ride_id, event_type, data):
    event = {
        'ride_id': ride_id,
        'type': event_type,
        'data': data,
        'timestamp': int(time.time() * 1000)
    }
    
    # Same ride_id always goes to the same partition
    producer.produce(
        topic='ride-events',
        key=str(ride_id).encode('utf-8'),
        value=json.dumps(event).encode('utf-8'),
        callback=delivery_report
    )
    producer.flush()

# All these events for ride_42 arrive in order
produce_ride_event(42, 'ride_requested', {'pickup': 'Main St'})
produce_ride_event(42, 'driver_assigned', {'driver_id': 99})
produce_ride_event(42, 'ride_started', {'lat': 40.7128, 'lon': -74.0060})

Keys don't have to be IDs. In an e-commerce system, you might key by user_id to ensure all purchases from one customer stay ordered, or by inventory_sku to guarantee stock updates don't race. Think about what ordering constraints your domain needs, then pick the key accordingly.

Commit Offsets After Processing

An offset is Kafka's way of tracking which messages a consumer group has processed. Consumer group 1 can be at offset 1000 while consumer group 2 is at offset 500 on the same topic. When a consumer crashes and restarts, it picks up from its last committed offset, not from the beginning.

Committing too early (before processing) means you lose messages on failure. Committing too late kills throughput. The safe pattern is: pull a batch, process it, then commit. If processing fails, don't commit and let another consumer try.

from confluent_kafka import Consumer, KafkaError
import json

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processor-v1',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False  # Manual commit for control
})

consumer.subscribe(['orders'])

def process_order(event):
    order_id = event['order_id']
    # Simulate work: write to database, call payment API, etc.
    print(f'Processing order {order_id}')
    # If this raises, we won't commit and will retry
    if event.get('amount') < 0:
        raise ValueError(f'Invalid amount for order {order_id}')
    return True

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        
        if msg is None:
            continue
        
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                print(f'Error: {msg.error()}')
                break
        
        try:
            event = json.loads(msg.value().decode('utf-8'))
            process_order(event)
            # Only commit after successful processing
            consumer.commit(asynchronous=False)
        except Exception as e:
            print(f'Failed to process message: {e}')
            # Don't commit; the message will be redelivered
            
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

Set enable.auto.commit: False to take manual control. After each successful process, call commit(asynchronous=False). The synchronous commit blocks until Kafka acknowledges, which is slower but safer for critical workflows like payments or inventory updates.

Idempotent Handlers for Redelivery

Kafka does not guarantee exactly-once delivery—it guarantees at-least-once. Messages can be redelivered if a consumer crashes after processing but before committing the offset. Your handlers must be idempotent: running them twice produces the same result as running them once.

In a trading system, if you process an order to buy 100 shares and the consumer crashes before committing, that order message will be reprocessed. If your handler isn't idempotent, you'll end up with 200 shares bought. That's a critical bug.

The standard pattern is to use a deduplication key—often the event's ID—stored in your database:

import json
from confluent_kafka import Consumer
from datetime import datetime

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'trade-processor',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False
})

consumer.subscribe(['trades'])

def process_trade(event, db_connection):
    trade_id = event['trade_id']
    symbol = event['symbol']
    quantity = event['quantity']
    
    # Check if we've already processed this trade
    cursor = db_connection.cursor()
    cursor.execute(
        'SELECT id FROM processed_trades WHERE trade_id = %s',
        (trade_id,)
    )
    
    if cursor.fetchone() is not None:
        print(f'Trade {trade_id} already processed, skipping')
        return True  # Idempotent: no-op on redelivery
    
    # Execute the trade
    cursor.execute(
        '''INSERT INTO positions (symbol, quantity, timestamp)
           VALUES (%s, %s, %s)''',
        (symbol, quantity, datetime.utcnow())
    )
    
    # Record that we've processed it
    cursor.execute(
        'INSERT INTO processed_trades (trade_id) VALUES (%s)',
        (trade_id,)
    )
    
    db_connection.commit()
    print(f'Trade {trade_id} executed')
    return True

# In your consumer loop:
try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None or msg.error():
            continue
        
        event = json.loads(msg.value().decode('utf-8'))
        try:
            process_trade(event, db_connection)
            consumer.commit(asynchronous=False)
        except Exception as e:
            print(f'Trade processing failed: {e}')
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

The processed_trades table acts as your deduplication store. On first delivery, you insert the trade and record the ID. On redelivery (if it happens), you check, find it's already processed, and skip. Your position balance stays correct.

Alternatively, if your event ID is globally unique (UUIDs are ideal for this), you can use database constraints to prevent duplicates automatically and catch the violation gracefully.

Common Mistakes

Not setting keys: Leads to random partitioning and lost ordering. Every message should have a key that represents the entity it belongs to.

Auto-committing after poll: When you enable auto-commit, messages are marked as processed before your handler runs. If processing fails, you've already lost the message. Always commit manually after successful processing.

Ignoring delivery callbacks: Producing a message doesn't mean it reached Kafka. Network hiccups, broker downtime, and full disks all cause silent failures. Always attach a callback and log failures.

Processing without idempotency guards: Assuming exactly-once delivery and not deduplicating. In practice, rebalances and crashes are common. Build your handlers to survive reprocessing.

Flushing synchronously on every message: Calling flush() after each produce blocks until the broker responds, tanking throughput. Batch your produces and flush periodically or when shutting down.

Using consumer groups without understanding offsets: Each consumer group maintains its own offset per partition. New groups start at the beginning or end depending on your config. Know what auto.offset.reset does in your environment.

Key Takeaways

• Use confluent-kafka, not kafka-python, for production workloads. It's faster, more reliable, and handles edge cases better.
• Always set a key on your messages to guarantee ordering per entity. Keys determine which partition a message lands in.
• Commit offsets manually after successful processing, not before. Set enable.auto.commit: False and call commit() explicitly.
• Make your event handlers idempotent by checking a deduplication table or using database constraints. Messages may be redelivered after crashes or rebalances.
• Attach delivery callbacks to producers to catch network and broker failures. Silent failures are worse than loud ones.
• Understand offset management: each consumer group tracks offsets independently. Know your auto.offset.reset policy and what it means for your workload.