tracker-best-practice.py

import json
import logging
import signal
import time
from typing import Any, Dict, Optional

from confluent_kafka import Consumer, KafkaError, KafkaException, Message

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger("orders-consumer")

RUNNING = True


def stop(*_args):
    global RUNNING
    RUNNING = False


signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)


def process_order(order: Dict[str, Any]) -> None:
    """
    Your business logic here.

    Best practice: make this idempotent (e.g., UPSERT by order_id),
    so reprocessing the same message doesn't corrupt data.
    """
    # Example "processing":
    log.info("Received order: %s x %s from %s", order.get("quantity"), order.get("item"), order.get("user"))
    # If something fails, raise an exception:
    # raise RuntimeError("DB write failed")


def safe_decode_json(msg: Message) -> Optional[Dict[str, Any]]:
    try:
        value = msg.value()
        if value is None:
            return None
        return json.loads(value.decode("utf-8"))
    except Exception:
        log.exception("Failed to decode/parse message: topic=%s partition=%s offset=%s",
                      msg.topic(), msg.partition(), msg.offset())
        return None


def on_assign(consumer: Consumer, partitions):
    log.info("Assigned partitions: %s", partitions)
    consumer.assign(partitions)


def on_revoke(consumer: Consumer, partitions):
    # Commit what we've already processed+stored before losing partitions
    log.info("Partitions revoked: %s. Committing stored offsets...", partitions)
    try:
        consumer.commit(asynchronous=False)
    except Exception:
        log.exception("Commit failed during revoke")


def main():
    consumer_config = {
        "bootstrap.servers": "localhost:9092",
        "group.id": "order-tracker",
        "auto.offset.reset": "earliest",

        # Best-practice defaults for production consumers:
        "enable.auto.commit": False,        # you control when offsets are committed
        "enable.auto.offset.store": False,  # you control when offsets are considered 'done'
        # Optional but often useful:
        # "max.poll.interval.ms": 300000,    # increase if processing can take long
        # "session.timeout.ms": 45000,
    }

    consumer = Consumer(consumer_config)
    consumer.subscribe(["orders"], on_assign=on_assign, on_revoke=on_revoke)

    log.info("Consumer started. group.id=%s", consumer_config["group.id"])

    commit_every_n = 200          # tune for throughput
    commit_every_seconds = 5.0    # tune for throughput
    processed_since_commit = 0
    last_commit_ts = time.time()

    try:
        while RUNNING:
            msg = consumer.poll(1.0)
            if msg is None:
                # periodic commit by time (optional)
                if processed_since_commit > 0 and (time.time() - last_commit_ts) >= commit_every_seconds:
                    consumer.commit(asynchronous=True)
                    processed_since_commit = 0
                    last_commit_ts = time.time()
                continue

            if msg.error():
                # Partition EOF is not a real error; skip it
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    continue
                raise KafkaException(msg.error())

            order = safe_decode_json(msg)
            if order is None:
                # Decide your policy: skip, send to DLQ, etc.
                # If you skip, DO NOT commit/store this message, or you'll permanently drop it.
                continue

            # 1) Process (side effects: DB write, API call, etc.)
            process_order(order)

            # 2) Only after success: store offset (marks "done" locally)
            consumer.store_offsets(message=msg)

            # 3) Commit in batches (good throughput)
            processed_since_commit += 1
            now = time.time()
            if processed_since_commit >= commit_every_n or (now - last_commit_ts) >= commit_every_seconds:
                consumer.commit(asynchronous=True)   # commits the stored offsets
                processed_since_commit = 0
                last_commit_ts = now

    except Exception:
        log.exception("Fatal error in consumer loop")
    finally:
        try:
            # Final sync commit to reduce duplicates after shutdown
            log.info("Shutting down: committing stored offsets...")
            consumer.commit(asynchronous=False)
        except Exception:
            log.exception("Final commit failed")
        consumer.close()
        log.info("Consumer closed.")


if __name__ == "__main__":
    main()