Faust and Kafka Consumer Patterns for Sensor Streams #

A single Python script polling a Kafka topic works fine until the moment it matters: a wildfire smoke event triples the ingest rate, one of your three consumer pods gets OOM-killed, and the survivors inherit its partitions mid-window. Without a disciplined consumer framework, that sequence produces duplicated air-quality averages, silently dropped late packets, and a stalled early-warning dashboard — exactly when the data is most consequential. faust-streaming (the maintained fork of Robinhood’s Faust) gives you the primitives to survive it: async agents bound to topics, partitioned Tables for fault-tolerant state, and explicit processing guarantees layered directly on the Kafka consumer group protocol. This guide covers how those pieces fit together for environmental telemetry, and it sits inside the broader Real-Time Stream Processing & Spatial Analytics pipeline as the ingestion-and-compute layer that reads from the broker and feeds windowed, spatially-enriched output downstream.

---

The three stages map directly to the sections below: a partitioned topic keyed for spatial locality, a consumer group whose agents own partitions and carry Table state through rebalances, and an idempotent sink whose write ordering defines your processing guarantee. Get the boundaries between them right and the failure modes above become recoverable rather than corrupting.

Prerequisites #

Confirm each of the following before wiring a faust agent into a live sensor pipeline. Every item maps to a concrete failure this layer cannot paper over.

• Python 3.11 with an event loop free of blocking calls. faust-streaming is fully async; a synchronous psycopg2 write inside an agent stalls the whole worker’s poll loop and triggers spurious rebalances when max.poll.interval.ms is exceeded. Wrap blocking sink I/O in loop.run_in_executor or use an async driver.
• Pinned versions: faust-streaming==0.11.1, confluent-kafka==2.3.0, pydantic==2.6.4. Pin these exactly — faust’s serializer/codec API and its rocksdb Table store are sensitive to minor version changes, and pydantic v2 model validation differs enough from v1 that mixing them silently drops fields. Add # python 3.11 · faust-streaming==0.11.1 · confluent-kafka==2.3.0 · pydantic==2.6.4 at the top of your app module.
• A running Kafka (or Redpanda) broker with the raw telemetry topic already created and partitioned. If you are still bridging devices onto the broker, stand up the upstream Kafka stream synchronization workflows first — a faust consumer group cannot compensate for a topic that never receives well-keyed messages.
• A partition key chosen for spatial locality: co-located sensors must share a partition so that per-region state and windowing execute within one agent instance rather than requiring cross-instance coordination. An H3 cell at resolution 5 or a stable device_id are the usual keys.
• Minimum record schema: device_id (string), sensor_ts (Unix milliseconds from the device clock), lat, lon (WGS 84 decimal degrees), and one or more metric fields. Records missing sensor_ts cannot participate in event-time windowing downstream and must be quarantined at validation, not silently consumed.
• An idempotent sink contract: your database target must support upsert on a natural key. At-least-once delivery guarantees you will occasionally reprocess an event; the sink is where that becomes harmless.

---

Step-by-Step Workflow #

Step 1 — Define the App and Record Schema #

The faust.App is the top-level object binding your consumer group to a broker. Its id becomes the Kafka consumer group id and the prefix for every internal topic (changelogs, repartition topics), so it must be stable across restarts — changing it silently starts a brand-new group that re-reads from the configured offset reset.

# python 3.11 · faust-streaming==0.11.1 · confluent-kafka==2.3.0 · pydantic==2.6.4
import faust
from pydantic import BaseModel, field_validator

class SensorReading(faust.Record, serializer="json"):
    device_id: str
    sensor_ts: int          # Unix ms from device clock
    lat: float
    lon: float
    pm25: float | None = None

class ValidatedReading(BaseModel):
    device_id: str
    sensor_ts: int
    lat: float
    lon: float
    pm25: float | None = None

    @field_validator("lat")
    @classmethod
    def _lat_range(cls, v: float) -> float:
        if not -90.0 <= v <= 90.0:
            raise ValueError(f"latitude {v} outside WGS84 range")
        return v

app = faust.App(
    "env-telemetry-consumer",          # == Kafka consumer group id — keep stable
    broker="kafka://broker:9092",
    store="rocksdb://",                 # local Table backing store
    topic_partitions=12,                # internal topics inherit this
    processing_guarantee="at_least_once",
)

raw_topic = app.topic("env.telemetry.raw", value_type=SensorReading)

The faust.Record handles wire serialization; the pydantic model does semantic validation on the way in. Keeping them separate lets you evolve validation rules without touching the codec. Complexity: app construction is O(1); topic declaration lazily registers metadata and costs nothing until the agent starts.

Step 2 — Declare an Agent Over the Topic #

An agent is an async coroutine decorated with @app.agent(topic). Declaring it is what tells faust to join the consumer group and request partition assignments for that topic. Each event flows through the agent’s async for loop exactly in per-partition offset order.

sink_topic = app.topic("env.telemetry.validated", value_type=SensorReading)

@app.agent(raw_topic)
async def process_readings(stream: faust.Stream[SensorReading]) -> None:
    async for reading in stream:
        try:
            clean = ValidatedReading(
                device_id=reading.device_id,
                sensor_ts=reading.sensor_ts,
                lat=reading.lat,
                lon=reading.lon,
                pm25=reading.pm25,
            )
        except ValueError:
            # Route bad records to a quarantine topic; never crash the agent.
            await quarantine_topic.send(value=reading)
            continue
        await sink_topic.send(key=clean.device_id, value=reading)

Two rules matter here. First, never let an exception escape the async for body — an uncaught error crashes the agent and forces a rebalance across the whole group. Catch, quarantine, and continue. Second, prefer stream.take(max_, within=) batching when your sink writes are dominated by round-trip latency; a batched agent amortizes the commit and cuts sink calls by the batch factor. Complexity: O(1) work per event, O(n) over the stream; memory is bounded by the stream buffer, not the topic size.

Step 3 — Choose a Processing Guarantee #

This is the decision that shapes everything downstream. faust-streaming defaults to at-least-once: it commits the consumed offset after your agent finishes processing an event. If the worker dies between the sink write and the commit, the event is replayed on restart. The alternative, processing_guarantee="exactly_once", wraps the produce-and-commit in a Kafka transaction so downstream Kafka topics and Table changelogs never see a duplicate — but it does nothing for an external database write, which is a side effect outside the transaction. The detailed agent build, including manual commit control, lives in building an at-least-once faust agent for sensor streams.

For environmental telemetry the pragmatic answer is almost always at-least-once plus an idempotent sink. Exactly-once is worth its overhead only when the terminal sink is itself another Kafka topic or a faust Table, where the transaction actually encloses the write.

# At-least-once with an idempotent DB write is the robust default.
async def upsert_reading(reading: SensorReading) -> None:
    # ON CONFLICT makes a replayed event overwrite, not duplicate.
    await db.execute(
        """
        INSERT INTO readings (device_id, sensor_ts, geom, pm25)
        VALUES ($1, $2, ST_SetSRID(ST_MakePoint($3, $4), 4326), $5)
        ON CONFLICT (device_id, sensor_ts) DO UPDATE SET pm25 = EXCLUDED.pm25
        """,
        reading.device_id, reading.sensor_ts, reading.lon, reading.lat, reading.pm25,
    )

Step 4 — Maintain Per-Key State in a Table #

When an agent needs memory — a rolling count, a last-seen timestamp for dedup, a running baseline — use a faust.Table, not a plain dict. A Table is partitioned by the same key as the stream and backed by a compacted Kafka changelog topic. When a partition moves during a rebalance, its Table entries are replayed onto the new owner from that changelog, so state follows the data. A plain dict, by contrast, is silently lost the moment a partition is reassigned.

# Table keyed by device_id; survives rebalances via its changelog topic.
last_seen = app.Table(
    "last_seen_ts",
    default=int,
    partitions=12,
)

@app.agent(raw_topic)
async def dedup_and_forward(stream: faust.Stream[SensorReading]) -> None:
    async for reading in stream.group_by(SensorReading.device_id):
        # Drop out-of-order duplicates using changelog-backed state.
        if reading.sensor_ts <= last_seen[reading.device_id]:
            continue
        last_seen[reading.device_id] = reading.sensor_ts
        await sink_topic.send(key=reading.device_id, value=reading)

stream.group_by() repartitions the stream so that all events for a device_id land on the same instance as its Table entry — without it, the Table lookup would read a partition the current instance may not own. The heavier rolling-state patterns this enables are covered in stateful stream processing patterns. Complexity: Table reads/writes are O(1) against the local rocksdb store; changelog replay on rebalance is O(k) in the number of keys on the moved partitions.

Step 5 — Tune Assignment and Backpressure #

Two forces destabilize a consumer group under load: partition assignment churn during rebalances, and buffer growth when the sink cannot keep pace. Address assignment with the cooperative-sticky assignor and static membership; address backpressure by bounding the stream buffer and the fetch size so a slow sink applies flow-control pressure back to the broker rather than exhausting memory. The full config surface for rebalancing is detailed in managing consumer group rebalancing for sensor topics.

app = faust.App(
    "env-telemetry-consumer",
    broker="kafka://broker:9092",
    processing_guarantee="at_least_once",
    stream_buffer_maxsize=4096,         # bounded in-flight events → backpressure
    broker_max_poll_records=500,        # cap fetch; lower during burst events
    consumer_max_fetch_size=1048576,    # 1 MiB per partition fetch
    broker_commit_every=1000,           # commit cadence (at-least-once)
    consumer_auto_offset_reset="earliest",
)

When stream_buffer_maxsize fills because a downstream write stalls, faust stops fetching from the broker — consumer lag rises visibly instead of memory silently exhausting. That is the signal to scale out or shed load, and it connects directly to the backpressure handling in Python streams strategies for adaptive polling and circuit breaking. Complexity: these are constant-space bounds; their whole purpose is to keep memory O(buffer) regardless of topic backlog.

---

Configuration and Tuning #

Consumer configuration is where sensor-network reality meets Kafka defaults, and the defaults assume a fast, reliable, homogeneous stream that environmental telemetry rarely is. The table below gives starting points by deployment profile.

Partition count is effectively permanent. You can add partitions to a live topic, but doing so remaps the key→partition hash and scatters a given device’s history across old and new partitions, breaking key-ordered state. Provision for two to three years of growth up front rather than resizing later.

Match session.timeout.ms to link quality, not to a copied default. A satellite-linked gateway whose heartbeats jitter by seconds will be repeatedly evicted and readmitted under the 10-second urban default, each cycle triggering a rebalance. Widen the timeout until eviction reflects genuine death, not transient latency.

---

Validation #

Before promoting a faust consumer to production, verify both correctness and stability under the conditions that actually break it.

• Offset-commit ordering: kill a worker with SIGKILL immediately after a sink write and confirm on restart that the replayed events produce no duplicate rows. If duplicates appear, the sink is not truly idempotent — the upsert key is wrong or incomplete.
• Rebalance survival: scale the group from two instances to three under steady load and confirm Table state is intact on the instance that inherits the moved partitions. Query the changelog topic’s compacted state and compare key counts before and after.
• Lag under burst: replay a 10× traffic spike and watch consumer group lag. With bounded stream_buffer_maxsize, lag should rise and then drain, never accompanied by climbing RSS. Growing memory instead of growing lag means a buffer is unbounded somewhere.
• Quarantine ratio: track the fraction of records routed to the quarantine topic. A sudden rise usually signals a firmware change altering the payload schema, not a consumer bug.

Expected shape: for a healthy 12-partition group with 3 instances, expect partitions distributed 4/4/4, per-partition lag under a few hundred messages at steady state, and zero duplicate (device_id, sensor_ts) rows in the sink across a controlled kill-and-recover test.

---

Failure Modes and Edge Cases #

Blocking I/O inside an agent stalls the poll loop. faust’s consumer heartbeat and message fetch share the same event loop as your agent. A synchronous database call that blocks for two seconds blocks heartbeats too; exceed session.timeout.ms and the broker evicts the instance, triggering a rebalance mid-batch. Always use async sink drivers, or offload blocking calls to an executor thread.

Rebalances mid-window duplicate aggregates. If an agent is accumulating a five-minute window in a Table and its partition moves before the window closes, the new owner replays from the last committed offset and re-accumulates. Without idempotent output keyed on the window boundary, you emit the window twice. Static membership plus the cooperative-sticky assignor minimizes this; idempotent sinks eliminate its consequences.

Unbounded group_by repartition topics. Every stream.group_by() writes to an internal repartition topic. If that key has high cardinality (e.g. grouping by raw sensor_ts instead of device_id), the repartition topic explodes and its retention silently fills disk. Group by stable, bounded keys only.

Changelog topic compaction lag. A Table’s fault tolerance depends on its changelog being compacted. If the broker’s log-cleaner is under-provisioned, changelog topics grow without bound and rebalance recovery — which replays the whole changelog — slows from seconds to minutes. Monitor changelog topic size relative to the Table’s live key count.

Silent codec mismatch on schema evolution. Adding a required field to SensorReading while old producers still emit the previous shape causes deserialization to drop or null the field with no error, because JSON codecs are permissive. Version your record schema and validate with the pydantic model on every event so drift surfaces as a quarantine spike rather than corrupt state.

---

Integration #

A faust consumer group is the compute hub between ingestion and analytics within the Real-Time Stream Processing & Spatial Analytics architecture. It connects outward in three directions:

1. Upstream it reads the well-keyed topic produced by the Kafka stream synchronization workflows bridge, which is responsible for getting device telemetry onto Kafka with a spatial partition key in the first place.
2. Within the compute layer its Table state feeds stateful stream processing patterns for rolling baselines and the windowed reductions in windowed aggregation for time-series, while its buffer bounds implement the flow control described in backpressure handling in Python streams.
3. Downstream it emits validated, deduplicated readings to an idempotent sink, carrying full provenance (device_id, sensor_ts, qc_flags) so audit and calibration reviews remain possible.

The two focused walkthroughs below build out the two hardest parts of this layer: implementing the agent with at-least-once crash recovery, and configuring the group so rebalances stay boring.

---

FAQ #

Does faust-streaming give me exactly-once processing out of the box?

No. faust-streaming defaults to at-least-once, committing offsets after an agent processes a message. It offers a processing_guarantee="exactly_once" mode that wraps produce-and-commit in a Kafka transaction, but that only covers writes back to Kafka topics and its own Table changelogs — writes to PostGIS or TimescaleDB are still external side effects you must make idempotent yourself.

How many partitions should my sensor telemetry topic have?

Provision partitions for peak parallelism, not current load: a consumer group can never run more useful agent instances than the topic has partitions. For most regional environmental networks 12–24 partitions on the raw telemetry topic is a good starting point, keyed by an H3 cell or device_id so co-located sensors land together. Over-partitioning wastes rebalance time; under-partitioning caps your throughput ceiling permanently because repartitioning a live topic reshuffles keys.

Why does my faust agent reprocess the same readings after a restart?

That is at-least-once semantics working as designed. If the process crashes after writing to the sink but before the offset commit lands, faust replays from the last committed offset on restart. The fix is not to chase exactly-once but to make the sink idempotent — upsert on (device_id, sensor_ts) so a replayed reading overwrites rather than duplicates.

What is the difference between a faust agent and a faust Table?

An agent is an async function that consumes a stream and processes each event; it is your consumer logic. A Table is a partitioned, fault-tolerant key-value store backed by a compacted Kafka changelog topic, used for state an agent accumulates — rolling counts, last-seen timestamps, dedup sets. When partitions move during a rebalance, the Table’s state for those keys is rebuilt on the new instance by replaying the changelog.

How do I stop a rebalance from creating duplicate windowed results?

Combine three settings: the cooperative-sticky assignor so most partitions keep their owner during a rebalance, static membership via group.instance.id so a pod restart does not trigger reassignment at all, and idempotent sinks keyed on the natural event key. With those in place a rebalance may replay a few uncommitted events but never produces a duplicate row downstream.

---

Related #

• Real-Time Stream Processing & Spatial Analytics — parent overview covering every stage from ingestion to spatial output
• Building an At-Least-Once Faust Agent for MQTT Sensor Streams — a self-contained agent with manual acking, idempotent sinks, and crash recovery
• Managing Consumer Group Rebalancing for Sensor Topics — sticky assignors, session timeouts, and static membership tuning
• Stateful Stream Processing Patterns — rolling baselines and checkpointing over faust Table state
• Backpressure Handling in Python Streams — bounded buffers and circuit breakers that pair with faust flow control