Consuming Messages with Python

A consumer reads records from a topic and does something with them. In a data pipeline that "something" is a write to a store such as ClickHouse. This article covers the poll loop, how consumer groups and rebalancing behave, the choices around committing offsets, and the batch-then-write pattern that makes the sink efficient and correct.

Mental model#

consumer.subscribe(["order-events"])   |loop:   msg = consumer.poll(timeout=1.0)   msg is None        -> nothing this cycle, loop again   msg.error()        -> handle or skip   otherwise          -> json.loads(msg.value()), process   |commit the offset    -> automatically every ~5s, or manually after the work   |group rebalances when a member joins or leaves:   partitions are revoked from some members and assigned to others

A consumer never works alone: it is a member of a consumer group. Kafka gives it a subset of the topic's partitions and can take them away and give them to another member at any time (a rebalance). Correct consumer code assumes its partition assignment can change.

Configure a consumer#

from confluent_kafka import Consumer
consumer = Consumer({    "bootstrap.servers": "localhost:9092",    "group.id": "warehouse-sink",    "auto.offset.reset": "earliest",    "enable.auto.commit": False,})
consumer.subscribe(["order-events"])

• group.id identifies the consumer group. Members of the same group share the partitions; different groups each get every record.
• auto.offset.reset applies only when the group has no committed offset for a partition yet: earliest starts at the beginning, latest at new records only. Once a committed offset exists, this setting is ignored.
• enable.auto.commit — True (the default) commits read offsets on a timer; False puts commits under your control. For a pipeline that writes downstream, False is usually right (see offset management below).

Use 127.0.0.1:9092 from the host, kafka:29092 from a container.

The poll loop#

poll(timeout) returns one record, or None if nothing arrived within the timeout. Always check for None and for msg.error() before touching the value:

import json
try:    while True:        msg = consumer.poll(timeout=1.0)
        if msg is None:            continue        if msg.error():            print(f"consumer error: {msg.error()}")            continue
        event = json.loads(msg.value().decode("utf-8"))        handle(event)except KeyboardInterrupt:    passfinally:    consumer.close()

poll() does more than return a record: it sends heartbeats to the group coordinator, drives rebalances, and (with auto-commit) commits offsets. If your code goes too long between poll() calls — longer than max.poll.interval.ms, 5 minutes by default — Kafka assumes the consumer is dead and rebalances its partitions away. Keep per-record work short, or process in bounded batches.

msg.value() is bytes; decode and parse it. msg.key(), msg.topic(), msg.partition(), and msg.offset() describe where the record came from.

Consumer groups and rebalancing#

Kafka assigns each partition of a subscribed topic to exactly one member of the group. When membership changes — a consumer starts, stops, or is considered dead — Kafka rebalances:

group "warehouse-sink", topic with 3 partitions
  one consumer:     C1 owns p0, p1, p2  add a consumer:   C1 owns p0, p1     C2 owns p2      (rebalance)  C1 crashes:       C2 owns p0, p1, p2                 (rebalance)

During a rebalance, partitions are revoked from current owners and assigned to new owners. Any in-progress, uncommitted work on a revoked partition will be re-read by whoever gets it next. That is fine if your processing is idempotent; if it is not, you can hook the transitions:

def on_assign(consumer, partitions):    print(f"assigned: {[ (p.topic, p.partition) for p in partitions ]}")
def on_revoke(consumer, partitions):    # last chance to flush and commit work for these partitions    flush_and_commit()
consumer.subscribe(["order-events"], on_assign=on_assign, on_revoke=on_revoke)

The practical rule: commit (and flush any downstream buffer) before you lose a partition, and design the sink so a replay of the last uncommitted batch is harmless.

Offset management#

The committed offset is a per-group, per-partition marker of "processed up to here", stored by Kafka in an internal topic. On restart, the consumer resumes from it.

Automatic commit (enable.auto.commit=True, the default): the client commits the latest polled offsets every auto.commit.interval.ms (5 seconds). Simple, but it commits records you may not have finished with — a crash can then skip them.

Manual commit (enable.auto.commit=False): you call commit() yourself, after the work is durable:

consumer.commit(asynchronous=False)          # commit the current position

asynchronous=False blocks until the broker acknowledges the commit; use it at batch boundaries where correctness matters more than a few milliseconds. asynchronous=True is fire-and-forget and fine for frequent, non-critical commits.

There is also store_offsets(msg) + enable.auto.commit=True + enable.auto.offset.store=False: you mark which offsets are safe to commit, and the client commits them on its timer. It is a middle ground — explicit about what is safe, without a synchronous commit per batch.

Batching before a downstream write#

Writing one row per record is slow for stores like ClickHouse, which want large inserts. Collect a batch, write it once, then commit:

BATCH_SIZE = 100batch = []
while True:    msg = consumer.poll(timeout=1.0)    if msg is None:        continue    if msg.error():        continue
    batch.append(json.loads(msg.value().decode("utf-8")))
    if len(batch) >= BATCH_SIZE:        write_batch(batch)                 # e.g. client.insert("raw.orders", ...)        consumer.commit(asynchronous=False)        batch.clear()

Order matters: write first, then commit. Trace the two failure points:

poll -> add to batch -> [A] write_batch -> [B] commit -> clear
crash at [A] (before the write):   offset not committed -> records re-read on restart -> written once. OK.
crash between [A] and [B] (written, not committed):   offset not committed -> records re-read -> written AGAIN.   -> at-least-once: the sink must tolerate the duplicate.
if you committed BEFORE the write and crashed:   offset committed -> records never written -> DATA LOSS.

So committing after the write gives at-least-once (possible duplicates); committing before gives at-most-once (possible loss). For data pipelines you want at-least-once plus a sink that de-duplicates.

Flush the partial batch on shutdown so the tail is not left uncommitted:

try:    ...  # the loop aboveexcept KeyboardInterrupt:    passfinally:    if batch:        write_batch(batch)        consumer.commit(asynchronous=False)    consumer.close()

Writing the batch to ClickHouse#

write_batch is the sink: it turns the list of decoded events into rows and loads them. Build the ClickHouse client once and return a write_batch closure that reuses it:

from clickhouse_connect import get_client
RAW_TABLE = "raw.orders"COLUMNS = ["order_id", "customer_id", "created_at", "amount"]

def make_clickhouse_sink():    client = get_client(        host="clickhouse",        port=8123,        username="default",        password="password",    )
    def write_batch(batch):        rows = []        for event in batch:            row = (                event["order_id"],                event["customer_id"],                event["created_at"],                event["amount"],            )            rows.append(row)
        client.insert(RAW_TABLE, data=rows, column_names=COLUMNS)
    return write_batch

Two details that are easy to get wrong:

• Append every row. Building row in the loop but forgetting rows.append(row) leaves rows empty, and client.insert() loads nothing with no error.
• Fix the column order. client.insert() maps data positionally to column_names, so the tuple must be built in the same order every time.

Because delivery is at-least-once, make the target de-duplicate on a business key — for example a ReplacingMergeTree keyed on order_id — so a replayed batch does not create duplicates. client.insert() and client.insert_df() are covered in Working with ClickHouse from Python.

Handling bad messages#

A record that cannot be processed — malformed JSON, a missing required field — will fail every time it is re-read, blocking the partition (a poison message). Decide the policy explicitly:

try:    event = json.loads(msg.value().decode("utf-8"))    validate(event)except (ValueError, KeyError) as exc:    log_bad_message(msg, exc)          # write it to a dead-letter topic or a table    continue                           # skip it; do NOT block the partition

Skipping silently loses data; blocking forever stalls the pipeline. Routing bad records to a dead-letter destination (another topic, or a quarantine table) keeps the main flow moving and preserves the record for investigation.

A complete consumer#

import jsonfrom confluent_kafka import Consumerfrom clickhouse_connect import get_client
TOPIC = "order-events"BATCH_SIZE = 100RAW_TABLE = "raw.orders"COLUMNS = ["order_id", "customer_id", "created_at", "amount"]

def make_clickhouse_sink():    client = get_client(        host="clickhouse", port=8123,        username="default", password="password",    )
    def write_batch(batch):        rows = []        for event in batch:            rows.append((                event["order_id"], event["customer_id"],                event["created_at"], event["amount"],            ))        client.insert(RAW_TABLE, data=rows, column_names=COLUMNS)
    return write_batch

def run(write_batch):    consumer = Consumer({        "bootstrap.servers": "localhost:9092",        "group.id": "warehouse-sink",        "auto.offset.reset": "earliest",        "enable.auto.commit": False,    })    consumer.subscribe([TOPIC])
    batch = []    try:        while True:            msg = consumer.poll(timeout=1.0)            if msg is None:                continue            if msg.error():                print(f"consumer error: {msg.error()}")                continue
            try:                event = json.loads(msg.value().decode("utf-8"))            except ValueError as exc:                print(f"bad message at {msg.partition()}/{msg.offset()}: {exc}")                continue
            batch.append(event)
            if len(batch) >= BATCH_SIZE:                write_batch(batch)                consumer.commit(asynchronous=False)                batch.clear()    except KeyboardInterrupt:        pass    finally:        if batch:            write_batch(batch)            consumer.commit(asynchronous=False)        consumer.close()

if __name__ == "__main__":    run(make_clickhouse_sink())

run() takes any write_batch(batch) callable, so the Kafka loop and the sink stay independent: the same consumer can write to ClickHouse, to a file, or to a test collector without changing the loop.

Monitoring: consumer lag#

Lag is log-end offset − committed offset per partition: how far the group is behind. It is the primary health signal for a streaming consumer.

• Steady low lag: the consumer keeps up.
• Lag that grows over time: the consumer cannot keep up; add consumers (up to the partition count), make write_batch faster, or increase BATCH_SIZE.
• Lag approaching the retention window: records will start expiring before they are read — an alert-now situation.

Check it with kafka-consumer-groups --describe --group <name> or in Kafka UI (see Running Kafka Locally).

Troubleshooting#

Common mistakes#

Committing before processing#

An early commit (including the auto-commit default) can mark records done that a crash then loses. Commit after the write.

Not handling None and error()#

poll() returns None on an idle cycle and an error message on a problem. Using msg.value() without checking crashes the loop.

Building rows but not collecting them#

A for loop that constructs row but never calls rows.append(row) leaves rows empty; client.insert() then loads zero rows with no error.

Never calling close()#

The group waits out the session timeout before reassigning the dead consumer's partitions. close() makes the handoff immediate.

More consumers than partitions#

Extra members of a group sit idle. Parallelism is capped at the partition count.

Blocking forever on a poison message#

Route unparseable records to a dead-letter destination instead of failing the loop or skipping silently.

Quick reference#

See also#