Message Broker Patterns for Capture Events

Within the Core Architecture for Shelf Analytics system, the message broker is the durable spine that carries every store capture event from the edge to the vision and scoring tiers, and its design decides whether the whole pipeline degrades gracefully or collapses under a chain-wide reset. A capture event is small — a signed pointer to a staged frame plus its store_id, fixture_id, and capture_timestamp — but the guarantees the broker makes about it are load-bearing: per-fixture ordering so consecutive-capture diffing works, at-least-once delivery so no compliance verdict is silently lost, backpressure so a burst of morning-reset traffic never overruns the GPU tier, and replay so a model upgrade can reprocess exactly the frames it should. Get the topic and partition design wrong and you get the two failure signatures every shelf analytics team eventually meets: a flagship store’s traffic starving the rest of the fleet from a single hot partition, and out-of-order captures on one fixture producing a phantom stockout that a category manager acts on at 7am. This page defines the capture-event contract the broker enforces, a runnable idempotent producer and consumer, the production configuration that tunes durability against throughput, the failure modes you will actually debug, and the partition-to-throughput math you size against.

Concept & Data Contract Jump to heading#

The broker’s job is to decouple capture from processing without losing the two properties the downstream tiers depend on: identity and order. Every message on the topic is a capture event — not the image itself, which lives in staging object storage, but a small signed envelope that points at it. The event is keyed on store_id so that all captures from one location land on the same partition, and it carries fixture_id and capture_timestamp so that per-fixture ordering and consecutive-capture diffing survive the trip. The payload is deliberately thin: a broker is a transport, not a data lake, and fattening the message with pixels would blow past sane message-size limits and cripple replay. A capture event on the wire looks like this:

{
  "schema_version": "shelf.capture.v1",
  "store_id": "US0421",
  "fixture_id": "BAY-014-SHELF-03",
  "capture_timestamp": "2026-06-28T07:14:22Z",
  "object_key": "s3://capture-staging/US0421/2026/06/28/BAY-014-SHELF-03/07-14-22.avif",
  "payload_sha256": "9f2b7c1e4a6d8f0b2c4e6a8d0f2b4c6e8a0d2f4b6c8e0a2d4f6b8c0e2a4d6f8b",
  "registry_revision": 184,
  "device_id": "handheld-77",
  "producer_seq": 40817
}

The partition key is store_id, and that choice is the single most consequential decision on this page. Kafka guarantees ordering only within a partition, so keying by store_id guarantees that every capture from one store is totally ordered relative to every other capture from that store — which is exactly the ordering per fixture the vision and scoring tiers assume, since a fixture belongs to exactly one store. Keying more finely, on fixture_id, would technically tighten the ordering guarantee to the fixture but would explode the key cardinality, scatter one store’s captures across many partitions, and destroy the ability to reason about a store as a unit during an outage or replay. Keying more coarsely, on region or banner, would collapse thousands of stores onto a handful of partitions and manufacture the hot-partition problem described below. store_id is the natural aggregate: coarse enough to bound cardinality, fine enough to preserve the ordering the pipeline actually needs.

The typed schema is enforced at both boundaries. The producer validates before publishing so a malformed event never reaches the log, and the consumer validates on the way out so a schema drift between producer and consumer fails loudly rather than corrupting a batch. A Pydantic contract makes the envelope self-describing and gives every downstream stage the same store_id, fixture_id, and capture_timestamp fields the rest of the Core Architecture for Shelf Analytics system agrees on:

from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator


class CaptureEvent(BaseModel):
    """The typed capture event carried on the broker. Pixels never travel here."""

    schema_version: str = Field(..., pattern=r"^shelf\.capture\.v\d+$")
    store_id: str = Field(..., pattern=r"^[A-Z]{2}\d{4}$")
    fixture_id: str
    capture_timestamp: datetime
    object_key: str = Field(..., min_length=8)
    payload_sha256: str = Field(..., min_length=64, max_length=64)
    registry_revision: int = Field(..., ge=0)
    producer_seq: int = Field(..., ge=0)

    @field_validator("capture_timestamp")
    @classmethod
    def must_be_utc_and_sane(cls, v: datetime) -> datetime:
        # Clock-skewed handhelds are a common source of out-of-order bugs.
        if v.tzinfo is None:
            raise ValueError("capture_timestamp must be timezone-aware UTC")
        if v.timestamp() > datetime.now(timezone.utc).timestamp() + 120:
            raise ValueError("capture_timestamp is implausibly in the future")
        return v

    def partition_key(self) -> bytes:
        """Key on store_id so a store's captures stay ordered on one partition."""
        return self.store_id.encode("utf-8")

Implementation Architecture Jump to heading#

The naive broker integration — fire a message and forget it, auto-commit offsets on a timer, catch nothing — fails in the two ways that matter most in retail: it silently duplicates events when a producer retries after a network blip, and it silently loses events when a consumer commits an offset before the work behind it succeeds. A correct integration makes the producer idempotent and keyed, makes the consumer commit manually only after the frame is durably handed off, and gives poison messages an explicit dead-letter path instead of an infinite retry loop. The example below uses confluent-kafka, chosen over a pure-Python client because its librdkafka core exposes the idempotent-producer machinery and the precise delivery-report callbacks this contract needs; aiokafka is the equivalent choice when the surrounding service is already asyncio-native.

An idempotent, keyed producer Jump to heading#

Idempotence (enable.idempotence=true) makes the broker deduplicate producer retries using the producer’s sequence numbers, so a redelivered event never appears twice on the log even when the network forces a retry. Keying every message on store_id routes it to the store’s partition. The delivery report is where a failed publish is caught and escalated rather than dropped:

from confluent_kafka import Producer, KafkaException


def build_producer(brokers: str) -> Producer:
    return Producer({
        "bootstrap.servers": brokers,
        "enable.idempotence": True,       # exactly-once semantics on the log
        "acks": "all",                    # all in-sync replicas must ack
        "max.in.flight.requests.per.connection": 5,  # safe with idempotence
        "compression.type": "zstd",
        "linger.ms": 20,                  # small batching window for throughput
        "retries": 2_147_483_647,         # retry indefinitely; idempotence dedupes
    })


def _on_delivery(err, msg) -> None:
    if err is not None:
        # Escalate: a lost capture is a lost compliance verdict.
        raise KafkaException(err)


def publish_capture(producer: Producer, topic: str, event: CaptureEvent) -> None:
    """Publish one validated capture event, keyed for per-store ordering."""
    producer.produce(
        topic=topic,
        key=event.partition_key(),
        value=event.model_dump_json().encode("utf-8"),
        on_delivery=_on_delivery,
    )
    producer.poll(0)  # serve delivery callbacks without blocking

A manual-commit consumer with a dead-letter path Jump to heading#

The consumer disables auto-commit and commits an offset only after the event has been validated and handed to the vision tier. A message that fails to parse or fails processing repeatedly is not retried forever — it is published to a dead-letter topic with its failure context and then acknowledged, so one poison event cannot stall its whole partition:

import json
from confluent_kafka import Consumer, Producer
from pydantic import ValidationError


def build_consumer(brokers: str, group_id: str) -> Consumer:
    return Consumer({
        "bootstrap.servers": brokers,
        "group.id": group_id,
        "enable.auto.commit": False,       # commit only after durable handoff
        "auto.offset.reset": "earliest",
        "max.poll.interval.ms": 300_000,   # allow long GPU batches without eviction
        "session.timeout.ms": 45_000,
    })


def run_consumer(
    consumer: Consumer, dlq_producer: Producer,
    topic: str, dlq_topic: str, handoff,
) -> None:
    consumer.subscribe([topic])
    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            continue  # transient; the client handles retriable partition errors
        try:
            event = CaptureEvent.model_validate_json(msg.value())
            handoff(event)                 # e.g. enqueue for batched detection
        except (ValidationError, ValueError) as exc:
            dlq_producer.produce(
                dlq_topic,
                key=msg.key(),
                value=msg.value(),
                headers=[("error", str(exc).encode("utf-8")[:400])],
            )
            dlq_producer.flush(5)
        # Commit AFTER success or after quarantine: at-least-once, never lost.
        consumer.commit(message=msg, asynchronous=False)

Two independent consumer groups read the same topic at their own offsets: the vision batching group that feeds Async Image Batching for High-Volume Stores, and a lighter realtime group that feeds Real-Time Compliance Alerting & Webhooks. Because consumer groups are isolated, the alerting group can run hot and near-tip while the vision group falls behind during a reset, and neither perturbs the other — the same broker serves two very different latency budgets from one durable log.

Production Configuration & Tuning Jump to heading#

The broker’s behavior under load is governed almost entirely by configuration, not code, and that configuration belongs in versioned files so durability and throughput can be reconciled without a redeploy. The topic itself is provisioned with enough partitions to carry peak throughput with headroom, replicated so a broker loss never loses a capture, and retained long enough that a replay window covers a full model-upgrade cycle:

topic:
  name: shelf.capture.v1
  partitions: 48            # sized to peak throughput; see Scaling section
  replication_factor: 3     # survive one broker loss with a quorum to spare
  configs:
    min.insync.replicas: 2  # with acks=all, tolerate one replica down
    retention.ms: 604800000 # 7 days: covers a full replay/upgrade cycle
    max.message.bytes: 65536      # envelopes are tiny; reject fat payloads
    cleanup.policy: delete

producer:
  enable.idempotence: true
  acks: all
  compression.type: zstd
  linger.ms: 20
  batch.size: 65536

consumer:
  vision_batching:
    group.id: capture-vision
    instances: 12           # <= partition count; one owns each partition
    enable.auto.commit: false
    max.poll.records: 200   # batch to the GPU memory envelope
    max.poll.interval.ms: 300000
  realtime_alerting:
    group.id: capture-alerting
    instances: 6
    enable.auto.commit: false
    max.poll.records: 50    # smaller batches keep alert latency low

The settings interlock. acks: all together with min.insync.replicas: 2 is the durability contract: a publish is acknowledged only once two of the three replicas hold it, so a single broker failure loses nothing, while still allowing the topic to accept writes when one replica is down for maintenance. Setting min.insync.replicas to 3 would be stricter but would block all writes the moment any replica lagged — a bad trade for a capture pipeline that must never refuse a store. Consumer group sizing follows the iron rule of Kafka: a partition is owned by at most one consumer in a group, so instances above the partition count sit idle. Twelve vision consumers against 48 partitions means each owns four partitions and there is room to scale to 48 before repartitioning. max.poll.records is tuned per group — large for the vision group so it fills GPU batches, small for the alerting group so a webhook fires within seconds — and max.poll.interval.ms is set generously on the vision group so a long detection batch does not trip a false liveness failure and trigger a rebalance. Retention of 7 days is the replay budget: it must exceed the longest interval between model upgrades so a rewind can always reach the events it needs to reprocess, which is the same replay guarantee the parent architecture relies on for Fallback Routing for Offline Store Scenarios.

Failure Modes & Debugging Workflow Jump to heading#

When the broker layer misbehaves, the symptom is almost always one of five recurring patterns. Work them in this order:

1. Hot partition from a flagship store. Symptom: one partition’s consumer lag climbs steadily while its siblings stay near tip, and end-to-end latency for one banner’s flagship location degrades while the rest of the fleet is healthy. Root cause: keying on store_id puts all of a single very-high-volume store’s captures on one partition, and that store out-produces what one consumer can drain. Fix: for the small set of flagship stores that individually saturate a partition, compose the key as store_id plus a bounded fixture bucket (for example store_id:hash(fixture_id) % 4) so the flagship’s load spreads across a few partitions while ordering is preserved per fixture — never per store for those keys. Confirm the fix by watching per-partition lag flatten. The partition-count math that prevents this in the first place is worked in Sizing Kafka Partitions for Store Capture Throughput.
2. Out-of-order captures on one fixture. Symptom: consecutive-capture diffing reports a fixture flipping between compliant and non-compliant on alternating captures, producing a phantom stockout. Root cause: events for one fixture reached different partitions — usually because a producer was misconfigured to key on device_id or a clock-skewed handheld stamped a capture_timestamp out of sequence. Fix: assert the partition key is store_id end to end, and reject future-dated timestamps at the producer as the contract above does; ordering is only ever guaranteed within a partition, so a fixture whose events span partitions can never be reordered downstream.
3. Consumer lag storms during chain-wide resets. Symptom: at a reset window every store captures at once, lag spikes across all partitions simultaneously, and the vision tier cannot keep pace. Root cause: capture rate briefly exceeds sustained consumer throughput — expected, not a defect. Fix: let the autoscaler add consumers up to the partition count while lag is the trigger signal (not CPU), and rely on the broker as the buffer it is; the log absorbs the burst and consumers drain it linearly once capacity is added. If lag routinely pins at the partition-count ceiling, the topic is under-partitioned and needs the sizing exercise, not more replicas.
4. Poison messages stalling a partition. Symptom: one partition’s offset stops advancing entirely while its consumer logs the same deserialization error in a tight loop. Root cause: a single malformed or unschema’d event that the consumer retries forever, blocking every event behind it in that partition. Fix: confirm the dead-letter path is wired as in the consumer above — validate, and on repeated failure publish to the DLQ and commit past the poison event rather than looping. Forensics on the quarantined events mirrors the dead-letter analysis in Error Handling in Computer Vision Pipelines.
5. Rebalance stalls from slow batches. Symptom: the whole consumer group pauses periodically, lag jumps on every partition at once, and the logs show repeated group rebalances. Root cause: a consumer took longer than max.poll.interval.ms to process a GPU batch, the coordinator judged it dead and evicted it, and the group rebalanced — often cascading as the reassigned load makes the next consumer slow too. Fix: raise max.poll.interval.ms above the worst-case batch time, lower max.poll.records so each poll’s work fits the interval, and prefer cooperative-sticky partition assignment so a rebalance moves only the affected partitions instead of stopping the world.

Keep a labeled incident log bucketed by these five causes. A monthly review of which bucket dominates tells you whether to repartition, fix a producer key, or raise a poll interval, and it is the only durable way to stop the same lag storm from recurring at every reset.

Scaling & Performance Benchmarks Jump to heading#

Partition count is the master throughput dial, because a topic’s sustained consumer throughput scales linearly with partitions up to the point where each partition has its own consumer. Size it from measured peak, not average: if a chain of 3000 stores each captures on average 2 events per second but a morning reset drives a 10x burst, peak ingress is roughly 60000 events per second. If a single consumer instance drains capture events — validate, hand off to a batch, commit — at about 1500 per second, sustaining that peak needs at least 40 consumers and therefore at least 40 partitions; provisioning 48 leaves headroom for a flagship hot-key split and for fleet growth without a disruptive repartition. Over-provisioning is not free — each partition costs open file handles, replication traffic, and rebalance time — so target roughly 2x peak headroom rather than an arbitrarily large count. The full sizing worksheet, including how to measure per-consumer drain rate and translate reset multipliers into partition counts, is the subject of Sizing Kafka Partitions for Store Capture Throughput.

Consumer concurrency is bounded by that partition count on the vision group, so once every partition has an owner the next lever is per-consumer batch efficiency, not more instances — which is why the vision group hands captures to the async batching tier rather than invoking the model per event. Set an explicit lag SLO and alert on it: for the realtime alerting group, keep p95 end-to-end lag under 10s so a compliance webhook fires while a store associate is still on the floor; for the vision batching group, a softer target of p95 < 90s from capture to a queued detection batch is appropriate during reset windows, tightening to p95 < 20s off-peak. Watch consumer lag as the truth signal, exactly as the parent architecture prescribes — rising lag means add consumers, sustained lag at the partition ceiling means repartition, and lag that will not drain even with idle partitions means the poison-message or rebalance path is stuck, not the throughput. Cost discipline mirrors the rest of the platform: the broker carries only tiny keyed envelopes, never frames, so at chain scale its bandwidth and storage are a rounding error against the GPU tier, and a 7-day retention on 65536-byte messages keeps the replay buffer cheap while still covering a full model-upgrade cycle.

Frequently Asked Questions Jump to heading#

Related Jump to heading#

• Core Architecture for Shelf Analytics — the parent system whose ingestion, resilience, and observability layers this broker spine sits inside
• Designing a Scalable Shelf Analytics Architecture — the sibling that sets the component-isolation and routing decisions this broker topology implements
• Fallback Routing for Offline Store Scenarios — how buffered offline captures replay back into this broker on reconnect
• Sizing Kafka Partitions for Store Capture Throughput — the child worksheet for turning peak capture rate into a partition count
• Async Image Batching for High-Volume Stores — the downstream vision consumer group that drains this topic into GPU batches
• Real-Time Compliance Alerting & Webhooks — a second consumer group that reuses this broker for low-latency alerting