Streaming Instrument Events into Kafka Topics #

This is the concrete producer-side task behind the decision to run Apache Kafka for live telemetry: taking a stream of instrument readings and publishing them into a Kafka topic so that each instrument’s events stay strictly ordered, no event is silently lost, and a retried write never creates a duplicate. It assumes you have already made the broker choice covered in Streaming Instrument Data: Choosing and Operating a Message Broker, that you are comfortable with Python 3.10+ and asyncio, and that you have a reachable Kafka broker (local or managed). The goal is a production async producer using aiokafka whose four design decisions — the partition key, the partition count, the delivery acks, and the idempotent-producer setting — are chosen deliberately rather than left at defaults that quietly drop or reorder scientific data.

The failure this prevents is subtle. A naive producer that omits a key spreads one instrument’s readings across every partition, destroying per-instrument order; one that uses acks=0 fires and forgets, losing events on any broker hiccup; one that retries without enable_idempotence duplicates events on transient errors. Each of these corrupts a time series without raising an obvious error at publish time. The producer below closes all three gaps.

The Core Design Decisions #

Four producer decisions determine whether the stream is correct. Every row below is a real setting with a concrete consequence; there are no placeholders. These are the values a production instrument producer actually ships with.

Setting enable_idempotence=True in aiokafka implies acks="all" and bounded in-flight requests, so the durability and dedup guarantees are consistent by construction. The partition key is the single most important choice: it is what converts Kafka’s per-partition ordering guarantee into the per-instrument ordering your time series depends on.

Two of these decisions deserve a sizing rule rather than a default. The partition count should be set to at least the number of instruments you expect to stream concurrently at peak, because a consumer group can assign at most one reader per partition — provision too few and your downstream fan-out is capped no matter how many consumer replicas you run, provision far too many and you pay for thousands of tiny segment files and slower rebalances. A count in the low tens is typical for a single-lab telemetry topic. The acks setting trades latency for durability: acks="all" waits for every in-sync replica and is the only safe choice for data you cannot regenerate, while acks=1 returns as soon as the partition leader has the record and silently loses it if that leader fails before replication. For irreplaceable instrument readings, the small added latency of acks="all" is never the wrong trade, which is why the idempotent producer forces it. Neither setting can be safely left implicit; both are chosen here against the shape of the workload.

Production Implementation #

The producer is an async context-managed object. It serializes each event to JSON, keys it by instrument id, and awaits the send so a broker-side failure surfaces as an exception you can route rather than a silent drop. The idempotent producer is enabled so the internal retries that ride out a transient broker error cannot duplicate an event.

from __future__ import annotations

import asyncio
import json
import logging
from datetime import datetime, timezone
from typing import Any

from aiokafka import AIOKafkaProducer
from aiokafka.errors import KafkaError
from pydantic import BaseModel, Field, field_validator

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("instrument.producer")

TOPIC = "instrument.telemetry.v1"
BOOTSTRAP = "localhost:9092"


class InstrumentEvent(BaseModel):
    """A single validated telemetry reading, published as one Kafka record."""

    instrument_id: str = Field(min_length=1, pattern=r"^[A-Za-z0-9_-]+$")
    sequence: int = Field(ge=0)            # monotonic per-instrument counter
    measured_at: datetime                  # timezone-aware UTC
    channel: str
    value: float

    @field_validator("measured_at")
    @classmethod
    def _require_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("measured_at must be timezone-aware")
        return v.astimezone(timezone.utc)


def _serialize_value(event: InstrumentEvent) -> bytes:
    """Deterministic JSON wire format; datetimes as ISO-8601 strings."""
    return event.model_dump_json().encode("utf-8")


def _serialize_key(instrument_id: str) -> bytes:
    """Partition key: identical bytes route an instrument's events to one partition."""
    return instrument_id.encode("utf-8")


class TelemetryPublisher:
    """Async wrapper around an idempotent aiokafka producer."""

    def __init__(self, bootstrap: str = BOOTSTRAP, topic: str = TOPIC) -> None:
        self._topic = topic
        self._producer = AIOKafkaProducer(
            bootstrap_servers=bootstrap,
            enable_idempotence=True,   # implies acks="all" + no duplicate retries
            acks="all",
            linger_ms=5,               # small batching window for throughput
            request_timeout_ms=30_000,
            value_serializer=_serialize_value,
            key_serializer=_serialize_key,
        )

    async def __aenter__(self) -> "TelemetryPublisher":
        await self._producer.start()
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self._producer.stop()   # flushes buffered records before closing

    async def publish(self, event: InstrumentEvent) -> None:
        """Publish one event, awaiting broker acknowledgement from all replicas."""
        try:
            metadata = await self._producer.send_and_wait(
                self._topic,
                key=event.instrument_id,     # key_serializer -> bytes
                value=event,                 # value_serializer -> bytes
            )
        except KafkaError as exc:
            # Durable send failed after internal retries: surface, do not swallow.
            logger.error("publish failed for %s seq=%d: %s",
                         event.instrument_id, event.sequence, exc)
            raise
        logger.info("published %s seq=%d -> partition=%d offset=%d",
                    event.instrument_id, event.sequence,
                    metadata.partition, metadata.offset)


async def stream_instrument(publisher: TelemetryPublisher,
                            readings: list[dict[str, Any]]) -> None:
    """Validate then publish a burst of raw readings for one or more instruments."""
    for raw in readings:
        event = InstrumentEvent.model_validate(raw)
        await publisher.publish(event)


async def main() -> None:
    readings = [
        {"instrument_id": "cyto-01", "sequence": 0,
         "measured_at": "2026-07-16T09:00:00Z", "channel": "FSC", "value": 1180.0},
        {"instrument_id": "cyto-01", "sequence": 1,
         "measured_at": "2026-07-16T09:00:01Z", "channel": "FSC", "value": 1192.5},
    ]
    async with TelemetryPublisher() as publisher:
        await stream_instrument(publisher, readings)


if __name__ == "__main__":
    asyncio.run(main())

Three details carry the guarantees. send_and_wait awaits the broker acknowledgement, so acks="all" means the call returns only after every in-sync replica has the record — a durable write, not a hopeful one. Keying by instrument_id pins each instrument to one partition, so its events are totally ordered by offset regardless of how many partitions the topic has. And enable_idempotence=True lets aiokafka retry a transiently failed send internally without risk of duplication, because the broker deduplicates on the producer id and per-partition sequence number. The Pydantic V2 model is the same validation discipline the ingestion boundary applies elsewhere: a malformed reading raises before it is ever serialized.

Verification #

You can assert the two properties that matter — deterministic serialization and correct partition keying — without a running broker, by exercising the serializers and the model directly. Then a single integration check against a local broker confirms an end-to-end round trip.

import json

import pytest
from pydantic import ValidationError


def test_key_is_stable_bytes() -> None:
    # The same instrument always serializes to identical key bytes -> same partition.
    assert _serialize_key("cyto-01") == _serialize_key("cyto-01") == b"cyto-01"


def test_value_roundtrips_to_json() -> None:
    event = InstrumentEvent.model_validate({
        "instrument_id": "seq-07", "sequence": 3,
        "measured_at": "2026-07-16T12:00:00Z", "channel": "Q30", "value": 0.94,
    })
    decoded = json.loads(_serialize_value(event))
    assert decoded["instrument_id"] == "seq-07"
    assert decoded["sequence"] == 3
    assert decoded["measured_at"].endswith("+00:00")   # UTC-normalized


def test_naive_timestamp_is_rejected() -> None:
    with pytest.raises(ValidationError):
        InstrumentEvent.model_validate({
            "instrument_id": "seq-07", "sequence": 0,
            "measured_at": "2026-07-16T12:00:00", "channel": "Q30", "value": 0.9,
        })

Run it with pytest -q. For an end-to-end smoke test against a local broker, publish two events and consume them back with a kafka-console-consumer or a short AIOKafkaConsumer subscribed to instrument.telemetry.v1 from offset zero; the two cyto-01 records must appear on the same partition in ascending sequence order.

Gotchas #

• Publishing without awaiting the send. Calling producer.send(...) and not awaiting the returned future (or not calling flush/stop) drops buffered events on shutdown. Fix: use send_and_wait, or await the futures and let the async context manager’s stop() flush on exit.
• Enabling retries without the idempotent producer. Plain retries on a transient error duplicate events and corrupt a monotonic series. Fix: set enable_idempotence=True, which makes aiokafka retries safe and forces acks="all".
• Repartitioning a live topic. Increasing partition count changes the key-to-partition mapping for future events, so an instrument can jump partitions and its global order across the boundary is no longer guaranteed. Fix: size partitions up front for peak fan-out, and treat any later increase as a planned migration.

Related Guides #

• Streaming Instrument Data: Choosing and Operating a Message Broker — the parent comparison that motivates Kafka’s log, partitioning, and delivery semantics used here.
• Async batch processing — the batch counterpart that consumes and reconciles these events downstream with the same idempotency guarantees.