Slotting Architecture · 10 min read

How to Stream Pick Events with Kafka Consumers

You have a WMS publishing every confirmed pick to a Kafka topic, and you need a consumer that turns those events into a live per-SKU velocity counter that re-slotting can read seconds after the pick happens. The trap is that a naive consumer — auto-commit on, += qty per message — double-counts on every rebalance and loses events on every crash. This page builds one focused, typed consumer that reads pick events, updates an idempotent rolling counter, and commits offsets manually in the correct order. It is the concrete Kafka implementation behind the Real-Time Streaming Velocity Updates guide, which sits inside the wider Velocity Data Ingestion & WMS Sync Pipelines architecture.

Prerequisites

Confirm each before pointing the consumer at a live topic:

  • Python 3.10+ — the code uses X | None unions, list[...] generics, and dataclass config.
  • confluent-kafka 2.3+ — the librdkafka-backed client used here; kafka-python works with the same loop shape if you prefer a pure-Python dependency.
  • A keyed topic — the WMS producer must key each message by sku_id so all events for a SKU are ordered on one partition. Without keying, no single consumer owns a SKU’s counter and the idempotency guarantees below break.
  • Read access to a consumer group — a group id scoped to velocity aggregation, with enable.auto.commit disabled so you control commit timing.
  • An idempotency key on the payload — a stable event_id per movement, so a redelivered message is a no-op rather than a double count.

Configuration Block

Every tunable lives in one externalized profile. The levers that decide correctness are enable.auto.commit (must be false) and commit_every_n (how many applied events between manual commits); the lever that decides freshness is window_seconds.

# kafka_velocity.yaml — one profile per facility
consumer:
  bootstrap_servers: "kafka-1:9092,kafka-2:9092"
  topic: "wms.inventory.movements"
  group_id: "velocity-aggregator"
  auto_offset_reset: "latest"      # cold start reads new events only
  enable_auto_commit: false        # MUST be false; we commit manually after the fold
  max_poll_interval_ms: 300000     # give the fold headroom before the broker evicts us
  window_seconds: 3600             # trailing window for the rolling counter (60 min)
  commit_every_n: 500              # applied events between manual offset commits
  poll_timeout_s: 1.0              # seconds to block on each poll
  dedup_capacity: 500000           # bounded set of recently-applied event ids
# Equivalent Python config dict consumed by the runner
KAFKA_VELOCITY = {
    "bootstrap_servers": "kafka-1:9092,kafka-2:9092",
    "topic": "wms.inventory.movements",
    "group_id": "velocity-aggregator",
    "auto_offset_reset": "latest",
    "enable_auto_commit": False,
    "max_poll_interval_ms": 300000,
    "window_seconds": 3600,
    "commit_every_n": 500,
    "poll_timeout_s": 1.0,
    "dedup_capacity": 500000,
}

Implementation

The consumer disables auto-commit, folds each event into a rolling per-SKU counter behind an idempotency set, and commits offsets only after a batch of applied events is durable. The rolling counter keeps a deque of (event_ts, qty) samples and evicts anything older than the window on every update, so rate(sku) is always the trailing-window pick quantity.

from __future__ import annotations

import json
import logging
import time
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta

from confluent_kafka import Consumer, KafkaError, Message

logger = logging.getLogger("velocity.kafka")


@dataclass
class RollingCounter:
    """Trailing-window pick quantity per SKU, updated incrementally."""
    window: timedelta
    samples: dict[str, deque[tuple[datetime, int]]] = field(default_factory=dict)
    totals: dict[str, int] = field(default_factory=dict)

    def apply(self, sku_id: str, ts: datetime, qty: int) -> None:
        dq = self.samples.setdefault(sku_id, deque())
        dq.append((ts, qty))
        self.totals[sku_id] = self.totals.get(sku_id, 0) + qty
        horizon = ts - self.window
        while dq and dq[0][0] < horizon:
            _, old = dq.popleft()
            self.totals[sku_id] -= old

    def rate(self, sku_id: str) -> int:
        return self.totals.get(sku_id, 0)


class VelocityConsumer:
    """Idempotent, manual-commit Kafka consumer for pick-event velocity."""

    def __init__(self, cfg: dict[str, object]) -> None:
        self.cfg = cfg
        self.counter = RollingCounter(window=timedelta(seconds=int(cfg["window_seconds"])))
        self._seen: dict[str, None] = {}
        self._dedup_cap = int(cfg["dedup_capacity"])
        self._since_commit = 0
        self._consumer = Consumer({
            "bootstrap.servers": cfg["bootstrap_servers"],
            "group.id": cfg["group_id"],
            "auto.offset.reset": cfg["auto_offset_reset"],
            "enable.auto.commit": cfg["enable_auto_commit"],
            "max.poll.interval.ms": cfg["max_poll_interval_ms"],
        })

    def _is_new(self, event_id: str) -> bool:
        if event_id in self._seen:
            return False
        self._seen[event_id] = None
        if len(self._seen) > self._dedup_cap:
            del self._seen[next(iter(self._seen))]
        return True

    def _fold(self, msg: Message) -> bool:
        """Apply one message idempotently; return True if it changed state."""
        payload = json.loads(msg.value())
        if payload.get("event_type") != "PICK" or payload.get("qty", 0) <= 0:
            return False
        if not self._is_new(payload["event_id"]):
            logger.debug("duplicate %s skipped", payload["event_id"])
            return False
        ts = datetime.fromisoformat(payload["event_ts"])
        self.counter.apply(payload["sku_id"], ts, int(payload["qty"]))
        return True

    def run(self, max_batches: int | None = None) -> None:
        """Poll, fold, and commit manually after every commit_every_n applied events."""
        self._consumer.subscribe([self.cfg["topic"]])
        commit_every = int(self.cfg["commit_every_n"])
        batches = 0
        try:
            while max_batches is None or batches < max_batches:
                msg = self._consumer.poll(float(self.cfg["poll_timeout_s"]))
                batches += 1
                if msg is None:
                    continue
                if msg.error():
                    if msg.error().code() != KafkaError._PARTITION_EOF:
                        logger.error("consume error: %s", msg.error())
                    continue
                if self._fold(msg):
                    self._since_commit += 1
                if self._since_commit >= commit_every:
                    self._consumer.commit(asynchronous=False)
                    logger.info("committed offsets after %d applied events",
                                self._since_commit)
                    self._since_commit = 0
        finally:
            self._consumer.commit(asynchronous=False)  # flush before leaving group
            self._consumer.close()
            logger.info("consumer closed; tracking %d SKUs", len(self.counter.totals))

Step-by-Step Walkthrough

  1. Disable auto-commit. The Consumer is built with enable.auto.commit set from enable_auto_commit (false). Auto-commit checkpoints offsets on a timer regardless of whether the fold succeeded, so a crash between commit and fold silently drops events. Manual commit ties the offset to the work.
  2. Fold before you count _since_commit. _fold parses the JSON, filters non-PICK and zero-qty messages, and runs the event_id through _is_new. Only a genuinely new, applied event increments _since_commit, so the commit cadence tracks real work, not raw message volume.
  3. Keep the dedup set bounded. _is_new inserts each event_id into an insertion-ordered dict and evicts the oldest once it passes dedup_capacity. This is what makes the counter idempotent: a message redelivered after a rebalance is recognized and skipped instead of double-counted.
  4. Update the rolling counter. RollingCounter.apply appends (event_ts, qty), adds to the running total, then evicts every sample older than window_seconds from the head of the deque. rate(sku_id) is therefore always the trailing-window quantity with no rescan.
  5. Commit after the fold, in order. Once commit_every_n applied events accumulate, commit(asynchronous=False) checkpoints synchronously. Because the commit happens strictly after the fold, an un-committed offset is simply redelivered on restart and the idempotency set absorbs the repeat — no event is lost and none is counted twice.
  6. Flush on the way out. The finally block commits and closes so a clean shutdown leaves the group at a known offset and the next start resumes exactly where the fold stopped.

Verification

Drive the counter and dedup logic directly, with no live broker, by feeding synthetic messages through _fold. A fake message object supplies value() and a null error().

import json
import logging
from datetime import datetime


class _FakeMsg:
    def __init__(self, payload: dict) -> None:
        self._raw = json.dumps(payload).encode()

    def value(self) -> bytes:
        return self._raw

    def error(self):
        return None


def test_idempotent_and_windowed() -> None:
    logging.basicConfig(level=logging.INFO)
    consumer = VelocityConsumer(KAFKA_VELOCITY | {"window_seconds": 3600})
    base = {"event_type": "PICK", "sku_id": "SKU1", "qty": 5,
            "event_ts": datetime(2025, 5, 20, 11, 0, 0).isoformat()}

    consumer._fold(_FakeMsg({**base, "event_id": "e1"}))
    consumer._fold(_FakeMsg({**base, "event_id": "e1"}))  # redelivery -> skipped
    consumer._fold(_FakeMsg({**base, "event_id": "e2", "qty": 3}))

    assert consumer.counter.rate("SKU1") == 8  # 5 + 3, e1 applied once
    print(f"SKU1 trailing-window rate = {consumer.counter.rate('SKU1')}")


test_idempotent_and_windowed()

Sample expected output:

INFO:velocity.kafka:... (no double count)
SKU1 trailing-window rate = 8

Common Pitfalls

  • Rebalances replaying uncommitted events. When the group rebalances, partitions you had not yet committed are reassigned and their events redelivered. Without the event_id dedup set every replayed pick inflates velocity. Keep the idempotency check in the hot path and never widen the commit cadence so far that a rebalance replays a huge batch.
  • Duplicate delivery treated as new demand. At-least-once delivery guarantees repeats, not uniqueness. A consumer that counts every message as a fresh pick will read a redelivery storm as a demand spike and thrash re-slotting. The dedup set is mandatory, not optional hardening.
  • Committing before the fold. Any commit that runs before the counter update turns a crash into lost events — the offset advanced but the work did not happen. Always fold first, then commit, so an un-committed offset is a safe redelivery rather than a hole in the counter.
  • Unbounded dedup or window state. A dedup set that never evicts and a deque that never trims both leak memory as the catalog grows. Cap the set with LRU eviction at dedup_capacity and evict expired samples on every apply, so steady-state memory is bounded by live SKU count, not total event volume.