Kafka Consumer Groups with Python for Scalable Streaming

Overview

Kafka consumer groups enable parallel processing of a topic by distributing partitions among consumers. Each partition is assigned to exactly one consumer in a group, providing horizontal scalability and ordered processing per partition. Below: creating consumer groups, managing offsets, handling rebalances, commit strategies, and achieving exactly-once semantics with Python and confluent-kafka.

When to Use This

• High-throughput event streaming (click streams, IoT data, financial transactions)
• Log aggregation and real-time analytics pipelines
• Event sourcing architectures where order per partition matters
• Stream processing with parallel consumers and partition-level ordering

Prerequisites

• Python 3.10+
• Kafka cluster (local or cloud, e.g., Confluent Cloud)
• confluent-kafka package

Solution

1. Basic Consumer Group

from confluent_kafka import Consumer, KafkaError
import json

def create_consumer(group_id: str, servers: str = 'localhost:9092'):
    conf = {
        'bootstrap.servers': servers,
        'group.id': group_id,
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,  # Manual commits
        'partition.assignment.strategy': 'cooperative-sticky',
    }
    return Consumer(conf)

def consume_messages(consumer: Consumer, topic: str):
    consumer.subscribe([topic])

    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
                print(f"Consumer error: {msg.error()}")
                continue

            # Process message
            key = msg.key().decode('utf-8') if msg.key() else None
            value = json.loads(msg.value().decode('utf-8'))

            print(f"Topic={msg.topic()}, Partition={msg.partition()}, "
                  f"Offset={msg.offset()}, Key={key}")

            process_event(value)

            # Manual commit after processing
            consumer.commit(msg)

    except KeyboardInterrupt:
        print("Stopping consumer...")
    finally:
        consumer.close()

consumer = create_consumer('order-processors')
consume_messages(consumer, 'orders')

2. Producer with Key-Based Partitioning

from confluent_kafka import Producer
import json

def create_producer(servers: str = 'localhost:9092'):
    return Producer({
        'bootstrap.servers': servers,
        'acks': 'all',
        'enable.idempotence': True,
    })

def delivery_report(err, msg):
    if err:
        print(f"Delivery failed: {err}")
    else:
        print(f"Delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")

def produce_event(producer: Producer, topic: str, key: str, value: dict):
    producer.produce(
        topic=topic,
        key=key.encode('utf-8'),  # Key determines partition
        value=json.dumps(value).encode('utf-8'),
        callback=delivery_report,
    )
    producer.poll(0)  # Serve delivery callbacks

# Key-based partitioning ensures same key always goes to same partition
producer = create_producer()
produce_event(producer, 'orders', 'user-123', {'orderId': 'o1', 'userId': 'user-123'})
produce_event(producer, 'orders', 'user-456', {'orderId': 'o2', 'userId': 'user-456'})
producer.flush()

3. Rebalance Handling

from confluent_kafka import Consumer, KafkaError, TopicPartition
import json

def on_assign(consumer, partitions):
    print(f"Partitions assigned: {partitions}")
    # Load committed offsets and seek to them
    for tp in partitions:
        # Could load from external store for exactly-once
        pass

def on_revoke(consumer, partitions):
    print(f"Partitions revoked: {partitions}")
    # Commit current offsets before losing ownership
    consumer.commit(asynchronous=False)

def on_lost(consumer, partitions):
    print(f"Partitions lost: {partitions}")
    # Partitions lost due to consumer failure — cleanup

def create_resilient_consumer(group_id: str):
    conf = {
        'bootstrap.servers': 'localhost:9092',
        'group.id': group_id,
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,
        'partition.assignment.strategy': 'cooperative-sticky',
        # Rebalance callbacks
        'on_assign': on_assign,
        'on_revoke': on_revoke,
        'on_lost': on_lost,
    }
    return Consumer(conf)

consumer = create_resilient_consumer('order-processors')
consumer.subscribe(['orders'])

4. Batch Processing with Manual Offset

from confluent_kafka import Consumer, KafkaError
import json
from collections import defaultdict

def consume_batch(consumer: Consumer, topic: str, batch_size: int = 100, timeout: float = 5.0):
    consumer.subscribe([topic])
    messages = []

    while True:
        msg = consumer.poll(timeout=timeout)

        if msg is None:
            if messages:
                process_batch(messages)
                consumer.commit(asynchronous=False)
                messages = []
            continue

        if msg.error():
            continue

        messages.append({
            'topic': msg.topic(),
            'partition': msg.partition(),
            'offset': msg.offset(),
            'key': msg.key().decode('utf-8') if msg.key() else None,
            'value': json.loads(msg.value().decode('utf-8')),
        })

        if len(messages) >= batch_size:
            process_batch(messages)
            consumer.commit(asynchronous=False)
            messages = []

def process_batch(messages: list):
    # Group by partition for ordered processing
    by_partition = defaultdict(list)
    for msg in messages:
        by_partition[(msg['topic'], msg['partition'])].append(msg)

    for (topic, partition), msgs in by_partition.items():
        print(f"Processing {len(msgs)} messages from {topic}[{partition}]")
        for msg in msgs:
            process_event(msg['value'])

5. Exactly-Once Semantics (Transactional)

from confluent_kafka import Consumer, Producer, KafkaError, KafkaException
import json

class ExactlyOnceProcessor:
    def __init__(self, input_topic: str, output_topic: str, group_id: str):
        self.consumer = Consumer({
            'bootstrap.servers': 'localhost:9092',
            'group.id': group_id,
            'auto.offset.reset': 'earliest',
            'enable.auto.commit': False,
            'isolation.level': 'read_committed',
        })

        self.producer = Producer({
            'bootstrap.servers': 'localhost:9092',
            'transactional.id': f'{group_id}-txn',
            'enable.idempotence': True,
        })

        self.input_topic = input_topic
        self.output_topic = output_topic

    def start(self):
        self.producer.init_transactions()
        self.consumer.subscribe([self.input_topic])

        try:
            while True:
                msg = self.consumer.poll(1.0)
                if msg is None or msg.error():
                    continue

                # Process within a transaction
                self.producer.begin_transaction()

                try:
                    result = process_event(json.loads(msg.value().decode('utf-8')))

                    # Produce output message
                    self.producer.produce(
                        self.output_topic,
                        value=json.dumps(result).encode('utf-8'),
                    )

                    # Commit consumer offset within the transaction
                    self.producer.send_offsets_to_transaction(
                        [TopicPartition(msg.topic(), msg.partition(), msg.offset() + 1)],
                        self.consumer.consumer_group_metadata(),
                    )

                    self.producer.commit_transaction()

                except Exception as e:
                    print(f"Transaction failed, aborting: {e}")
                    self.producer.abort_transaction()

        finally:
            self.consumer.close()
            self.producer.flush()

6. Running Multiple Consumers in a Group

# Run multiple instances — Kafka assigns partitions automatically
# With 6 partitions and 3 consumers, each gets 2 partitions

# consumer_worker.py
import sys
from confluent_kafka import Consumer

def start_worker(worker_id: str, group_id: str, topic: str):
    consumer = Consumer({
        'bootstrap.servers': 'localhost:9092',
        'group.id': group_id,
        'auto.offset.reset': 'earliest',
        'enable.auto.commit': False,
        'partition.assignment.strategy': 'cooperative-sticky',
    })

    consumer.subscribe([topic])
    print(f"[Worker {worker_id}] Subscribed to {topic}")

    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None or msg.error():
                continue

            value = json.loads(msg.value().decode('utf-8'))
            print(f"[Worker {worker_id}] P={msg.partition()} O={msg.offset()} — {value.get('id')}")
            process_event(value)
            consumer.commit(msg)
    finally:
        consumer.close()

worker_id = sys.argv[1]
start_worker(worker_id, 'order-processors', 'orders')

# Run in separate terminals:
# python consumer_worker.py worker-1
# python consumer_worker.py worker-2
# python consumer_worker.py worker-3

How It Works

1. Consumer groups: Consumers in the same group share partitions. Each partition is assigned to exactly one consumer. Adding consumers scales throughput up to the number of partitions.
2. Key-based partitioning: The producer hashes the message key to determine the partition. Same key always goes to the same partition, preserving order for that key.
3. Offset management: Each consumer tracks its position (offset) in each partition. Commits persist the offset to Kafka’s __consumer_offsets topic. Manual commits give control over when offsets are saved.
4. Rebalance: When consumers join or leave the group, partitions are reassigned. Cooperative-sticky strategy minimizes disruption by only moving partitions that need to move.
5. Exactly-once: Transactions tie the consumer offset commit and producer send into one atomic operation. If the transaction fails, both are rolled back — no duplicates, no data loss.

Variants

Windowed Aggregation with Kafka Streams

# Use confluent-kafka-python's stream processing or Faust
# pip install faust-streaming

import faust

app = faust.App('order-aggregator', broker='kafka://localhost:9092')
orders_topic = app.topic('orders', value_type=dict)
order_counts = app.Table('order_counts', default=int)

@app.agent(orders_topic)
async def count_orders(stream):
    async for order in stream:
        user_id = order['userId']
        order_counts[user_id] += 1
        print(f"User {user_id}: {order_counts[user_id]} orders")

Schema Registry with Avro

from confluent_kafka import Consumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer

schema_registry = SchemaRegistryClient({'url': 'http://localhost:8081'})
avro_deserializer = AvroDeserializer(schema_registry)

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'avro-consumers',
})

consumer.subscribe(['orders-avro'])

msg = consumer.poll(1.0)
if msg:
    value = avro_deserializer(msg.value(), None)
    print(f"Deserialized: {value}")

Multiple Topics

# Subscribe to multiple topics with pattern
consumer.subscribe(['orders', 'payments', 'shipments'])

# Or use regex pattern
import re
consumer.subscribe(['orders.*'], on_assign=on_assign)

Best Practices

• For a deeper guide, see Consume Kafka Topics with Spring Boot Stream Listeners.

• Use cooperative-sticky assignment: The default range strategy causes stop-the-world rebalances. Cooperative-sticky only moves affected partitions, reducing disruption.

• Disable enable.auto.commit: Auto-commit can commit offsets before processing completes. Use manual commits after successful processing.

• Use key-based partitioning for order: Messages with the same key (e.g., user ID) go to the same partition, preserving order. Without keys, messages are distributed round-robin.

• Set isolation.level=read_committed: Only read committed transactions. Without it, consumers may see uncommitted (potentially rolled-back) messages.

• Handle rebalances gracefully: Implement on_revoke to commit offsets before losing partition ownership. Otherwise, you may reprocess messages.

• Monitor consumer lag: Use kafka-consumer-groups.sh or Burrow to monitor lag (difference between latest offset and committed offset). High lag means consumers can’t keep up.

Common Mistakes

• More consumers than partitions: Extra consumers sit idle. If you have 6 partitions, only 6 consumers can actively consume. Scale partitions before scaling consumers.
• Auto-commit with slow processing: Auto-commit runs every 5 seconds by default. If processing takes longer, offsets are committed before processing completes — data loss on crash.
• Not handling rebalance: During rebalance, partitions are revoked. If you don’t commit before revocation, you reprocess messages after the rebalance.
• Blocking in poll loop: Long-running processing blocks poll(), triggering session timeout and rebalance. Use pause/resume or async processing.
• Not setting auto.offset.reset: If no committed offset exists, this setting determines where to start. earliest reads from the beginning, latest only new messages. Choose based on your use case.