Producer Design

What "production-ready" means for a producer, concretely: it does not silently drop a record on a transient network blip. It does not block your application's request thread for hundreds of milliseconds on every single send. It tells you, unambiguously, when a write failed instead of leaving a future or callback nobody checked. It shuts down without abandoning messages still sitting in its internal buffer. And it exposes metrics that let you notice degradation before a customer does.

from confluent_kafka import Producer

producer = Producer({
    'bootstrap.servers': 'broker-1:9092,broker-2:9092',
    'acks': 'all',
    'enable.idempotence': True,
})

# ANTI-PATTERN: blocking on every single send
for order_event in order_events:
    future = producer.produce(
        topic='orders.events',
        key=order_event['order_id'],
        value=serialize(order_event),
    )
    producer.flush()  # blocks until THIS record is fully acknowledged
    # next iteration cannot start until the flush above returns
    # effective throughput: roughly one record per network round trip

# Benchmark: 100,000 order events, single producer, acks=all,
# 3-broker cluster, ~2ms average network round trip

# Pattern A — flush() after every produce() call:
#   100,000 records x ~2ms round trip = ~200,000ms = ~200 seconds
#   effective throughput: ~500 records/second

# Pattern B — async produce() with batching (linger.ms=10, batch.size=32768):
#   100,000 records arrive far faster than network round trips can drain them
#   producer accumulates ~2,000-record batches, sends ~50 requests total
#   50 requests x ~5ms (larger batch, still one round trip) = ~250ms
#   effective throughput: >100,000 records/second

# The difference is not "a bit slower." It is roughly 200x.
# Both patterns use the exact same acks=all durability guarantee.

from confluent_kafka import Producer
import logging

logger = logging.getLogger(__name__)

producer = Producer({
    'bootstrap.servers': 'broker-1:9092,broker-2:9092',
    'acks': 'all',
    'enable.idempotence': True,
    'linger.ms': 10,
    'batch.size': 32768,
    'compression.type': 'lz4',
})

def delivery_callback(err, msg):
    if err is not None:
        logger.error(
            f"Delivery failed for key={msg.key()} "
            f"topic={msg.topic()} error={err}"
        )
        # decide here: alert, dead-letter, or fail the upstream request
    else:
        logger.debug(
            f"Delivered key={msg.key()} to "
            f"{msg.topic()}[{msg.partition()}]@{msg.offset()}"
        )

for order_event in order_events:
    producer.produce(
        topic='orders.events',
        key=order_event['order_id'],
        value=serialize(order_event),
        on_delivery=delivery_callback,
    )
    # poll(0) services delivery callbacks without blocking the send loop
    producer.poll(0)

# after the loop, drain everything still in flight before moving on
producer.flush(timeout=30)

from confluent_kafka import KafkaError
import logging

logger = logging.getLogger(__name__)

# Errors where the message itself is permanently unsendable —
# retrying with the same payload will never succeed
NON_RETRIABLE_CODES = {
    KafkaError.MSG_SIZE_TOO_LARGE,
    KafkaError.INVALID_MSG_SIZE,
    KafkaError.TOPIC_AUTHORIZATION_FAILED,
    KafkaError.UNKNOWN_TOPIC_OR_PARTITION,
}

def make_delivery_callback(dlq_producer, dlq_topic):
    def delivery_callback(err, msg):
        if err is None:
            return  # success — nothing to do

        error_code = err.code()

        if error_code in NON_RETRIABLE_CODES:
            logger.error(
                f"Non-retriable delivery failure, routing to DLQ: "
                f"key={msg.key()} error={err}"
            )
            dlq_producer.produce(
                topic=dlq_topic,
                key=msg.key(),
                value=msg.value(),
                headers=[('failure_reason', str(err).encode())],
            )
            dlq_producer.poll(0)
        else:
            # producer already exhausted its own retry budget for this message
            logger.critical(
                f"Retriable error exhausted internal retries — "
                f"likely broker/network instability: key={msg.key()} error={err}"
            )
            # this is an operational alert, not a per-message fix
            emit_metric('producer.delivery_failure.retriable_exhausted')

    return delivery_callback

import json
from decimal import Decimal

def serialize_order_event(event: dict) -> bytes:
    try:
        return json.dumps(event, default=_json_default).encode('utf-8')
    except (TypeError, ValueError) as exc:
        logger.error(
            f"Serialization failed for order_id={event.get('order_id')}: {exc}"
        )
        emit_metric('producer.serialization_failure')
        raise  # let the caller decide: skip, dead-letter, or fail the request

def _json_default(obj):
    if isinstance(obj, Decimal):
        return str(obj)
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")

producer.produce(
    topic='orders.events',
    key=order_id.encode('utf-8'),
    value=serialize_order_event(event),
    headers=[
        ('trace_id', trace_context.trace_id.encode('utf-8')),
        ('schema_version', b'3'),
        ('producing_service', b'checkout-service'),
    ],
)

# a downstream consumer can inspect headers without deserializing
# the full payload -- useful for routing or filtering before the
# more expensive deserialization step even runs
def route_by_schema_version(msg):
    headers = dict(msg.headers() or [])
    version = headers.get('schema_version', b'1').decode()
    if version == '1':
        return parse_legacy_order_event(msg.value())
    return parse_order_event_v3(msg.value())

import json
import logging
from decimal import Decimal
from typing import Optional

from confluent_kafka import Producer, KafkaError

logger = logging.getLogger(__name__)

NON_RETRIABLE_CODES = {
    KafkaError.MSG_SIZE_TOO_LARGE,
    KafkaError.INVALID_MSG_SIZE,
    KafkaError.TOPIC_AUTHORIZATION_FAILED,
    KafkaError.UNKNOWN_TOPIC_OR_PARTITION,
}


class OrderEventProducer:
    """
    Production producer for order lifecycle events.
    Durability posture: acks=all, idempotent, unbounded retries within
    a two-minute delivery budget. Optimized for throughput via batching,
    not per-record latency -- appropriate for an events stream, not a
    synchronous request/response path.
    """

    def __init__(self, bootstrap_servers: str, topic: str, dlq_topic: str):
        self.topic = topic
        self.dlq_topic = dlq_topic
        self._sent_count = 0
        self._failed_count = 0

        self.producer = Producer({
            'bootstrap.servers': bootstrap_servers,
            'acks': 'all',
            'enable.idempotence': True,
            'max.in.flight.requests.per.connection': 5,
            'retries': 2147483647,
            'delivery.timeout.ms': 120000,
            'request.timeout.ms': 30000,
            'linger.ms': 10,
            'batch.size': 32768,
            'compression.type': 'lz4',
            'client.id': 'order-events-producer',
        })

        # separate producer instance for the DLQ -- keeps DLQ writes
        # from competing with primary-topic batching and simplifies
        # reasoning about each producer's delivery guarantees independently
        self.dlq_producer = Producer({
            'bootstrap.servers': bootstrap_servers,
            'acks': 'all',
            'enable.idempotence': True,
            'client.id': 'order-events-producer-dlq',
        })

    @staticmethod
    def _json_default(obj):
        if isinstance(obj, Decimal):
            return str(obj)
        raise TypeError(f"Object of type {type(obj)} is not JSON serializable")

    def _serialize(self, event: dict) -> Optional[bytes]:
        try:
            return json.dumps(event, default=self._json_default).encode('utf-8')
        except (TypeError, ValueError) as exc:
            logger.error(
                f"Serialization failed for order_id={event.get('order_id')}: {exc}"
            )
            self._failed_count += 1
            return None

    def send_order_event(self, event: dict) -> bool:
        """
        Queue an order event for async delivery. Returns False immediately
        if the event could not even be serialized -- callers should treat
        that as a hard failure for this specific event, not retry blindly.
        Actual delivery success/failure is reported later via the
        delivery callback, not by this method's return value.
        """
        value = self._serialize(event)
        if value is None:
            return False

        order_id = event.get('order_id', '')

        try:
            self.producer.produce(
                topic=self.topic,
                key=order_id.encode('utf-8') if order_id else None,
                value=value,
                on_delivery=self._delivery_callback,
            )
        except BufferError:
            # local producer queue is full -- this means we are producing
            # faster than the broker can absorb, even with batching.
            # Block briefly to drain, then retry once.
            logger.warning("Producer local queue full, blocking to drain")
            self.producer.poll(1.0)
            self.producer.produce(
                topic=self.topic,
                key=order_id.encode('utf-8') if order_id else None,
                value=value,
                on_delivery=self._delivery_callback,
            )

        # service any callbacks that are ready without blocking the caller
        self.producer.poll(0)
        return True

    def _delivery_callback(self, err, msg):
        if err is None:
            self._sent_count += 1
            logger.debug(
                f"Delivered order_id={msg.key()} to "
                f"{msg.topic()}[{msg.partition()}]@{msg.offset()}"
            )
            return

        self._failed_count += 1
        error_code = err.code()

        if error_code in NON_RETRIABLE_CODES:
            logger.error(
                f"Non-retriable failure, routing to DLQ: "
                f"key={msg.key()} error={err}"
            )
            self._send_to_dlq(msg, str(err))
        else:
            logger.critical(
                f"Delivery failed after exhausting internal retries "
                f"(likely broker/network instability): "
                f"key={msg.key()} error={err}"
            )
            self._send_to_dlq(msg, str(err))

    def _send_to_dlq(self, msg, failure_reason: str):
        try:
            self.dlq_producer.produce(
                topic=self.dlq_topic,
                key=msg.key(),
                value=msg.value(),
                headers=[
                    ('failure_reason', failure_reason.encode('utf-8')),
                    ('source_topic', self.topic.encode('utf-8')),
                ],
            )
            self.dlq_producer.poll(0)
        except Exception as exc:
            # if even the DLQ write fails, this is the last line of defense --
            # log at the highest severity available, this needs a human now
            logger.critical(
                f"DLQ write itself failed for key={msg.key()}: {exc}"
            )
            emit_metric('producer.dlq_write_failure')

    def close(self, timeout: float = 30.0):
        """
        Flush every in-flight and buffered message before the process
        exits. Called from a signal handler or application shutdown hook --
        never let the process exit while messages are still buffered
        in the producer's local queue, or they are lost silently.
        """
        logger.info(
            f"Shutting down producer: {self._sent_count} sent, "
            f"{self._failed_count} failed. Flushing remaining messages..."
        )
        remaining = self.producer.flush(timeout=timeout)
        if remaining > 0:
            logger.critical(
                f"{remaining} messages still unflushed after {timeout}s "
                f"timeout -- these are LOST. Investigate broker health."
            )
            emit_metric('producer.shutdown_data_loss', value=remaining)

        self.dlq_producer.flush(timeout=timeout)
        logger.info("Producer shutdown complete.")

import signal
import sys

producer = OrderEventProducer(
    bootstrap_servers='broker-1:9092,broker-2:9092',
    topic='orders.events',
    dlq_topic='orders.events.dlq',
)

def handle_shutdown(signum, frame):
    logger.info(f"Received signal {signum}, beginning graceful shutdown")
    producer.close(timeout=25.0)  # leave headroom under the orchestrator's kill timeout
    sys.exit(0)

signal.signal(signal.SIGTERM, handle_shutdown)
signal.signal(signal.SIGINT, handle_shutdown)

def _delivery_callback(self, err, msg):
    if err is None:
        self._sent_count += 1
        emit_metric('producer.delivery.success')
        return

    self._failed_count += 1
    emit_metric('producer.delivery.failure', tags={'error_code': str(err.code())})

    if err.code() in NON_RETRIABLE_CODES:
        self._send_to_dlq(msg, str(err))
    else:
        emit_metric('producer.delivery.retriable_exhausted')
        self._send_to_dlq(msg, str(err))

import time
import subprocess

def test_producer_survives_leader_failure(docker_compose_cluster):
    producer = OrderEventProducer(
        bootstrap_servers='localhost:9092,localhost:9093,localhost:9094',
        topic='test.orders.events',
        dlq_topic='test.orders.events.dlq',
    )

    delivered = []
    failed = []

    def tracking_callback(err, msg):
        if err is None:
            delivered.append(msg)
        else:
            failed.append((msg, err))

    # send a batch, then kill the current leader mid-flight
    for i in range(1000):
        producer.producer.produce(
            topic='test.orders.events',
            key=str(i).encode(),
            value=json.dumps({'order_id': str(i)}).encode(),
            on_delivery=tracking_callback,
        )
        if i == 500:
            subprocess.run(['docker', 'kill', 'test-broker-leader'])
        producer.producer.poll(0)

    producer.producer.flush(timeout=60)

    # every record must be accounted for -- either delivered or explicitly
    # reported as failed. Silence is the only unacceptable outcome.
    assert len(delivered) + len(failed) == 1000
    assert len(delivered) >= 500  # everything before the kill should have landed

import time

def benchmark_producer_throughput(producer, record_count=200_000):
    start = time.monotonic()
    sent = 0

    def count_callback(err, msg):
        nonlocal sent
        if err is None:
            sent += 1

    for i in range(record_count):
        producer.producer.produce(
            topic='orders.events',
            key=f"order-{i}".encode(),
            value=make_realistic_order_payload(i),  # real field shapes, not {"x": 1}
            on_delivery=count_callback,
        )
        producer.producer.poll(0)

    producer.producer.flush(timeout=60)
    elapsed = time.monotonic() - start

    throughput = sent / elapsed
    print(f"Sent {sent}/{record_count} records in {elapsed:.2f}s "
          f"({throughput:,.0f} records/sec)")
    return throughput

def test_shutdown_flushes_all_buffered_messages():
    producer = OrderEventProducer(
        bootstrap_servers='localhost:9092',
        topic='test.orders.events',
        dlq_topic='test.orders.events.dlq',
    )

    for i in range(5000):
        producer.send_order_event({'order_id': str(i), 'status': 'created'})
    # deliberately do NOT wait for a natural flush -- simulate a shutdown
    # signal arriving immediately after a burst of sends

    producer.close(timeout=30.0)

    # a correct close() logs and alerts on a nonzero remaining count;
    # this test asserts the common case where the timeout is sufficient
    assert producer._sent_count + producer._failed_count == 5000

At Affirm: the payments platform team is reviewing a new producer for loan-installment events before it ships. The reviewer's first question, straight from Part 09's checklist, is not about business logic at all — it's "show me the delivery callback and the shutdown handler." The original PR had neither: send() was called with no callback, and there was no signal handling to flush on deploy. Both gaps are the kind that pass every unit test and only surface as silent data loss weeks later during a routine rolling deploy.

At Instacart: a batch-ingestion producer for shopper-location pings is bottlenecked far below expected throughput. Following Part 03's benchmark numbers, an engineer checks the code and finds producer.flush() called after every singleproduce() call — a leftover from an early prototype that nobody removed once real traffic arrived. Switching to async sends with a delivery callback and a 10mslinger.ms takes throughput from roughly 800 records/second to over 60,000/second with no change to the durability guarantee.

At Brex: a transaction-event producer starts throwing BufferErrorduring a traffic spike from a large customer's batch upload. Per Part 06's worked example, the on-call engineer recognizes this as the local producer buffer filling faster than the broker can absorb it — not a broker outage. The immediate fix is the brief-block-and-retry pattern from the worked producer class; the longer-term fix, tracked as a follow-up, is raisingbuffer.memory and revisiting partition count on the target topic.