Message Brokers and Queues — How They Work Internally

The three guarantees a broker makes that direct connections cannot:

Temporal decoupling — the consumer does not need to be running when the producer sends. A Razorpay payment event written at 3 AM is still there for the analytics consumer that starts processing at 6 AM. The broker held it safely in between.

Spatial decoupling — the producer does not know the consumer's address. It knows only the broker's address and the topic or queue name. New consumers can be added with zero changes to the producer.

Rate decoupling — the producer and consumer can run at completely different speeds. The broker absorbs the difference. If Swiggy's order service produces 50,000 events per second during the dinner rush but the fraud detection consumer can only process 10,000 per second, the broker buffers 40,000 events per second of lag without the producer slowing down or losing data.

# Queue behaviour:
# Producer sends 5 messages: M1 M2 M3 M4 M5
# Two consumers A and B both connected to the queue

# Queue state after all messages arrive:
# [M1] [M2] [M3] [M4] [M5]

# Consumer A pulls M1 → queue state: [M2] [M3] [M4] [M5]   (M1 is gone)
# Consumer B pulls M2 → queue state: [M3] [M4] [M5]         (M2 is gone)
# Consumer A pulls M3 → queue state: [M4] [M5]
# Consumer B pulls M4 → queue state: [M5]
# Consumer A pulls M5 → queue state: []

# Result: M1 M3 M5 processed by A | M2 M4 processed by B
# No message was delivered twice. No message was skipped.
# Neither consumer can go back and re-read M1. It is gone.

# Real use case: Flipkart order fulfilment
# 10,000 orders sitting in a queue
# 50 fulfilment worker threads all pulling from the same queue
# Each order is processed by exactly one worker — no double fulfilment

# Topic behaviour:
# Producer sends 3 messages: M1 M2 M3
# Three subscribers: notification-service, inventory-service, analytics-service

# Topic log (append-only, nothing is removed):
# offset 0: M1
# offset 1: M2
# offset 2: M3

# notification-service committed offset: 3 (has read all three)
# inventory-service committed offset:    2 (has read M1 and M2, M3 is pending)
# analytics-service committed offset:    0 (has read nothing yet, is behind)

# Neither inventory-service nor analytics-service being slow affects anyone else
# notification-service cannot "take" messages away from the others
# analytics-service can reset to offset 0 and replay all history

# If a 4th service joins now, it can also start from offset 0
# and read M1 M2 M3 — even though they were written days ago
# The topic held them. The producer wrote them once. Everyone gets them.

# On the broker's filesystem, one partition looks like this:
/data/kafka/freshmart.orders-3/          ← partition 3 of freshmart.orders topic
    00000000000000000000.log             ← segment starting at offset 0
    00000000000000000000.index           ← sparse offset → byte-position index
    00000000000000000000.timeindex       ← timestamp → offset index
    00000000000000985432.log             ← segment starting at offset 985432
    00000000000000985432.index
    00000000000000985432.timeindex
    00000000000001847291.log             ← active segment (currently being written)
    00000000000001847291.index
    00000000000001847291.timeindex

# The .log file: raw bytes of sequentially appended messages
# The .index file: sparse index mapping offset → byte position in the .log file
#   (not every offset is indexed — every Nth offset, configurable)
#   When a consumer requests offset 985500, the broker:
#   1. Binary searches the .index file for the largest indexed offset ≤ 985500
#   2. Seeks to that byte position in the .log file
#   3. Scans forward to offset 985500
#   This is O(log N) for the index lookup + O(small scan) — very fast

# The .timeindex file: maps timestamps to offsets
#   Used when a consumer says "start from 3 hours ago" instead of a specific offset
#   Broker translates the timestamp to an offset using this index

# freshmart.orders partition 0 — replicated across 3 brokers (replication factor = 3)

# Broker 1 (leader for partition 0):
#   log: [offset 0] [offset 1] ... [offset 10,000]   ← latest
#   ISR: [broker-1, broker-2, broker-3]               ← all three are in sync

# Broker 2 (follower):
#   log: [offset 0] [offset 1] ... [offset 9,998]     ← 2 messages behind leader
#   (currently fetching, will catch up in milliseconds)

# Broker 3 (follower):
#   log: [offset 0] [offset 1] ... [offset 10,000]    ← fully in sync

# Producer writes offset 10,001 to broker-1 (leader)
# With acks=all: producer waits until broker-1, broker-2, AND broker-3 all confirm
# → offset 10,001 is now on 3 machines
# → broker-1 crashes → broker-2 or broker-3 becomes new leader
# → offset 10,001 is NOT lost

# With acks=1: producer only waits for broker-1 to confirm
# → broker-1 crashes BEFORE broker-2 and broker-3 replicate offset 10,001
# → offset 10,001 is LOST — producer thinks it was written, broker says it wasn't
# → this is called unclean leader election + data loss

# The ISR list contract:
# acks=all means "wait until all replicas in the current ISR confirm"
# If broker-3 goes offline (removed from ISR), acks=all only waits for broker-1 + broker-2
# min.insync.replicas setting: refuse writes if ISR size drops below this number
# Recommended: replication.factor=3, min.insync.replicas=2
# → tolerates 1 broker failure without data loss
# → refuses writes if 2 brokers are down (rather than silently risking data loss)

# freshmart.products — compacted topic
# Tracks current product price. Key = product_id. Value = current price.

# Log BEFORE compaction (chronological, by offset):
# offset 0:  key=P1001  value={"name":"Toor Dal 1kg",  "price_paise": 18900}
# offset 1:  key=P1002  value={"name":"Basmati 5kg",   "price_paise": 67500}
# offset 2:  key=P1001  value={"name":"Toor Dal 1kg",  "price_paise": 19500}  ← price updated
# offset 3:  key=P1003  value={"name":"Sunflower Oil", "price_paise": 23400}
# offset 4:  key=P1002  value={"name":"Basmati 5kg",   "price_paise": 69000}  ← price updated
# offset 5:  key=P1001  value={"name":"Toor Dal 1kg",  "price_paise": 21000}  ← price updated again

# Log AFTER compaction:
# offset 3:  key=P1003  value={"name":"Sunflower Oil", "price_paise": 23400}   ← only version
# offset 4:  key=P1002  value={"name":"Basmati 5kg",   "price_paise": 69000}   ← latest
# offset 5:  key=P1001  value={"name":"Toor Dal 1kg",  "price_paise": 21000}   ← latest

# offsets 0, 1, 2 were deleted because P1001 and P1002 have newer versions

# What a new consumer reading from offset 0 sees:
# → starts from the earliest surviving offset (3)
# → sees the latest value for every key that has ever existed
# → has a complete and current view of all products — a "changelog"

# Tombstones — how to "delete" a key from a compacted topic:
# Produce a message with key=P1001 and value=null
# This is a tombstone. Compaction will eventually remove it.
# After the tombstone is compacted away, P1001 is gone from the log.

# ── Break 1: Producer retries with multiple in-flight requests ──────────────

# Producer config: max.in.flight.requests.per.connection = 5 (default)
# Producer sends batch B1 to the leader. Network hiccup — no acknowledgement.
# Producer sends batch B2 while waiting for B1's ack.
# B2 arrives and is written successfully (offset 100).
# B1 is retried and also arrives successfully (offset 101).
# Now B2 (later batch) has a lower offset than B1 (earlier batch).
# Consumers see B2 before B1. Ordering violated.

# Fix: set max.in.flight.requests.per.connection=1 (serialises requests)
# Better fix: enable.idempotence=true (Kafka handles deduplication + ordering)
# enable.idempotence=true forces max.in.flight=5 to be safe using sequence numbers

# ── Break 2: Consumer reading multiple partitions, merging by arrival order ──

# Topic freshmart.orders has 4 partitions.
# Consumer reads from all 4 and processes events in the order they arrive.
# Partition 0: order at 14:23:11 — arrives at consumer at 14:23:12
# Partition 2: order at 14:23:09 — arrives at consumer at 14:23:13 (1 sec delay)
# Consumer sees the 14:23:11 order BEFORE the 14:23:09 order.
# In processing time order, everything was fine.
# In event time order (which matters for business logic), it was wrong.

# Fix: do not assume cross-partition arrival order reflects event time order.
# If business logic requires event time ordering across partitions,
# buffer events and sort by event_time before processing — at the cost of latency.

# ── Break 3: Leader failover during write ────────────────────────────────────

# Producer sends message M to partition leader (broker-1).
# broker-1 writes M to its local log but crashes before replicating to followers.
# (This only happens with acks=1 or acks=0 — with acks=all, this cannot happen)
# broker-2 is elected new leader. Its log does not contain M.
# Producer retries M to broker-2 (new leader). M is written at a new, higher offset.
# If there were messages after M that DID replicate before the crash,
# M now appears AFTER those messages. Ordering is violated for any consumer
# that already read those messages.

# Fix: acks=all + min.insync.replicas=2. Non-negotiable for ordered data.

# Producer rate:  50,000 events/second
# Consumer rate:  45,000 events/second
# Net lag growth: 5,000 events/second

# After 1 hour:  5,000 × 3,600 = 18,000,000 events of lag
# After 1 day:   5,000 × 86,400 = 432,000,000 events of lag

# If retention is 7 days and average event size is 500 bytes:
# Total data in retention: 50,000 × 86,400 × 7 × 500 = ~15 TB

# With 432M events of lag and consumer rate of 45k/sec:
# Time to catch up (if producer stops): 432M / 45k = ~2.7 hours
# But the producer doesn't stop — the consumer can NEVER catch up
# without either increasing consumer throughput or decreasing producer rate.

# Lag metric to alert on:
# ALERT if lag > 100,000 AND lag is growing (slope > 0 over last 5 minutes)
# This catches the "slow leak" before it becomes a crisis

# Solutions to backpressure (in order of preference):
# 1. Scale out consumer group — add more consumers (up to partition count)
# 2. Optimise consumer processing — reduce per-event processing time
# 3. Increase partition count — allows more consumer parallelism (requires repartitioning)
# 4. Throttle the producer at the source (last resort — affects upstream systems)

import json
import logging
from typing import Callable

logger = logging.getLogger(__name__)

MAX_RETRIES = 3

def process_with_dlq(
    event: dict,
    process_fn: Callable,
    dlq_producer,
    dlq_topic: str,
):
    """
    Attempt to process an event. On repeated failure, route to DLQ.
    Caller must commit offset AFTER this function returns — whether success or DLQ.
    """
    last_exception = None

    for attempt in range(1, MAX_RETRIES + 1):
        try:
            process_fn(event)
            return  # Success — caller will commit offset
        except Exception as exc:
            last_exception = exc
            logger.warning(
                f"Processing failed (attempt {attempt}/{MAX_RETRIES}): "
                f"event_id={event.get('event_id')} error={exc}"
            )

    # All retries exhausted — send to DLQ
    dlq_event = {
        "original_event":   event,
        "failed_at":        "2026-03-20T14:23:11Z",  # use datetime.utcnow().isoformat()
        "error_message":    str(last_exception),
        "error_type":       type(last_exception).__name__,
        "retry_count":      MAX_RETRIES,
        "source_topic":     "freshmart.orders",
        "source_partition": event.get("_partition"),
        "source_offset":    event.get("_offset"),
    }

    dlq_producer.produce(
        topic=dlq_topic,
        key=event.get("partition_key"),
        value=json.dumps(dlq_event).encode(),
    )
    dlq_producer.flush()

    logger.error(
        f"Event sent to DLQ: event_id={event.get('event_id')} "
        f"dlq_topic={dlq_topic}"
    )
    # Return normally — caller commits offset, processing continues

# DLQ topic naming convention:
# source_topic + ".dlq"
# freshmart.orders → freshmart.orders.dlq
# freshmart.payments → freshmart.payments.dlq

# DLQ monitoring:
# Alert: DLQ message count > 0 (any failure needs attention)
# Dashboard: DLQ message rate (failures per minute)
# Runbook: how to inspect, fix, and replay DLQ messages

# Without idempotent producer (enable.idempotence=false):

# Producer sends batch [M1, M2, M3] with sequence numbers [101, 102, 103]
# Network delivers batch. Broker writes M1 M2 M3. Sends ack.
# Network drops the ack. Producer never receives it.
# Producer retries: sends [M1, M2, M3] again.
# Broker writes M1 M2 M3 AGAIN at new offsets.
# Partition now has: M1 M2 M3 M1 M2 M3 — DUPLICATES

# With idempotent producer (enable.idempotence=true):

# Producer sends [M1, M2, M3] with PID=5001, sequence=[101, 102, 103]
# Broker writes them. Sends ack. Network drops ack.
# Producer retries: sends [M1, M2, M3] again, same PID=5001, same sequences
# Broker checks: "PID 5001 already wrote sequence 101 to this partition"
# Broker discards M1 M2 M3 silently. Sends success ack.
# Partition has: M1 M2 M3 — exactly once, no duplicates

# The broker maintains a per-(PID, partition) window of 5 sequence numbers
# Sequences outside this window (too old) are rejected — not deduplicated
# This means idempotent producer only deduplicates retries within one producer session
# Producer restart = new PID = no deduplication for messages from the old session
# → This is why idempotent producer alone is not enough for crash recovery
# → You also need consumer-side idempotency (as covered in Module 40)

# Read from freshmart.orders, transform, write to freshmart.orders.enriched
# All in one transaction — either all succeed or none

from confluent_kafka import Producer, Consumer, KafkaError

# Producer configured with transactional.id
producer = Producer({
    'bootstrap.servers': 'broker:9092',
    'transactional.id': 'order-enrichment-service-1',  # unique per producer instance
    'enable.idempotence': True,  # required for transactions
})
producer.init_transactions()

consumer = Consumer({
    'bootstrap.servers': 'broker:9092',
    'group.id': 'order-enrichment-group',
    'isolation.level': 'read_committed',  # CRITICAL — only read committed messages
    'enable.auto.commit': False,           # CRITICAL — we commit inside the transaction
})
consumer.subscribe(['freshmart.orders'])

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

    try:
        event = json.loads(msg.value())
        enriched = enrich_order(event)  # your transformation logic

        producer.begin_transaction()

        # Write enriched event to output topic
        producer.produce('freshmart.orders.enriched', value=json.dumps(enriched))

        # Commit consumer offset INSIDE the transaction
        # This is what makes it atomic — offset moves only when write succeeds
        offsets = [{
            'topic': msg.topic(),
            'partition': msg.partition(),
            'offset': msg.offset() + 1,
        }]
        producer.send_offsets_to_transaction(offsets, consumer.consumer_group_metadata())

        producer.commit_transaction()
        # ↑ Either both the write AND the offset commit happened, or neither did

    except Exception as e:
        producer.abort_transaction()
        # Consumer offset was NOT committed — event will be reprocessed
        logger.error(f"Transaction aborted: {e}")

At a fintech (Razorpay / PhonePe): A production incident — the payment reconciliation job is showing duplicate transactions in its output. You are asked to investigate. You check the producer configuration: enable.idempotence=falseand retries=3. Every network blip causes duplicate messages. The fix is two lines of config change. But before you can make that change, you need to understand why idempotence works, what sequence numbers the broker tracks, and why it's safe to enable without changing application logic. This module is the prerequisite for that conversation.

At an e-commerce company (Nykaa / Myntra):You are asked to design a new notification system. Orders topic already exists. Inventory, fraud, loyalty, and analytics services all need to react to new orders. The architecture decision is: do we use a queue (one service, work distribution) or a topic (multiple services, each sees every order)? You immediately know the answer — topic-based pub-sub, because four different services need the same event and none should "consume" it away from the others. You also know to set replication factor 3 with min.insync.replicas 2 and acks=all because orders are financial data. These are not guesses — they follow directly from understanding the concepts.

In a system design interview: "How would you handle a message that keeps failing in your streaming pipeline?" The wrong answer is "add more retries." The right answer covers: exponential backoff with a retry limit, routing to a DLQ topic after N failures, committing the offset so downstream processing continues, alerting on DLQ depth, building a replay mechanism, and the operational runbook for investigating and replaying DLQ messages. Every one of those points is in this module.