Exactly-Once Sensor Ingestion with Kafka Transactions in Python

A crash halfway through a read-process-write loop is the failure that quietly corrupts environmental archives: the consumer read a batch of sensor readings, the sink wrote some of them, and then the process died before committing offsets — so on restart the same readings are reprocessed and a PM2.5 hourly mean is computed over duplicated observations. Kafka transactions close this window by committing the output messages and the consumed offsets as a single atomic unit, giving exactly-once semantics between Kafka topics; pairing that with an idempotent PostGIS upsert extends the guarantee to your storage sink for effectively-once end-to-end delivery. This page builds a self-contained transactional producer/consumer with the confluent-kafka transactional API, and it is the durability-focused part of the broader Kafka stream synchronization workflows for environmental sensor streams.

Why Transactions, Not Just Idempotent Producers

Enabling enable.idempotence on a producer stops the broker from writing duplicate copies of the same message during retries — valuable, but it only covers the producer-to-broker hop. It does nothing about the offsets your consumer has read. The real hazard in ingestion is the coupling between three actions that must all succeed or all fail together: consuming an input reading, producing a derived record (a normalized, geolocated row on an output topic), and advancing the consumer offset past that input. If offsets advance but the output write is lost, data vanishes; if the output write commits but offsets do not, the reading is reprocessed on restart.

Transactions bind the produce and the offset-commit into one atomic operation via send_offsets_to_transaction. The consumer’s offsets are written to the internal __consumer_offsets topic inside the same transaction as the output messages, so commit_transaction either publishes both or, on abort_transaction, discards both. The transactional.id gives the broker a stable identity to fence zombie instances: when a restarted producer calls init_transactions, the broker bumps the producer epoch and refuses commits from the old, possibly-still-alive process. The external PostGIS sink sits outside this boundary, which is why it must be idempotent — the same reasoning that makes downstream consumers in Faust Kafka consumer patterns rely on natural-key upserts rather than blind inserts.

It helps to be precise about what “exactly-once” does and does not promise, because the phrase is routinely oversold. Kafka guarantees exactly-once within Kafka: for a read-process-write loop whose input, output, and offsets all live in Kafka, no input is processed twice and no output is duplicated, even across arbitrary crashes and rebalances. The moment a side effect leaves Kafka — an HTTP call, a file write, a PostGIS insert — it is outside the transaction and the guarantee weakens to “at-least-once, made idempotent.” That is why the industry phrase for the end-to-end property is effectively-once: the Kafka layer ensures each reading is handled once, and the idempotent sink absorbs the rare replay that the unavoidable two-phase gap between the database commit and the Kafka commit can produce. Designing for effectively-once means accepting that a message may be processed more than once and engineering the sink so the result is identical regardless.

The alternative many teams reach for first — track processed message ids in a set, or write offsets to the database alongside the data — reinvents the transaction badly. Storing offsets in PostGIS in the same DB transaction as the rows does give you a consistent “what have I written” checkpoint, but it couples your offset management to your sink and breaks the moment you add a second output topic. Kafka transactions keep offset ownership where it belongs, in Kafka, and let the sink stay a dumb idempotent upsert. That separation is what makes the pattern scale to multiple output topics and multiple sinks without a combinatorial explosion of consistency logic.

Production-Ready Implementation

The loop below consumes raw sensor readings, transforms them, produces normalized rows to an output topic, records consumed offsets inside the transaction, and upserts to PostGIS idempotently. Auto-commit is disabled; offsets flow only through the transaction.

# python 3.11 · confluent-kafka==2.3.0 · psycopg2-binary==2.9.9
from __future__ import annotations

import json

import psycopg2
from confluent_kafka import Consumer, Producer, TopicPartition, KafkaError


def build_clients(bootstrap: str, group_id: str, txn_id: str) -> tuple[Consumer, Producer]:
    """Create a read_committed consumer (auto-commit off) and a transactional producer."""
    consumer = Consumer({
        "bootstrap.servers": bootstrap,
        "group.id": group_id,
        "enable.auto.commit": False,        # offsets commit via the transaction only
        "auto.offset.reset": "earliest",
        "isolation.level": "read_committed",
    })
    producer = Producer({
        "bootstrap.servers": bootstrap,
        "transactional.id": txn_id,         # STABLE across restarts — enables fencing
        "enable.idempotence": True,
    })
    return consumer, producer


def upsert_reading(cur, row: dict) -> None:
    """Idempotent PostGIS sink: dedupe on (dev_eui, received_at)."""
    cur.execute(
        """
        INSERT INTO sensor_readings (dev_eui, received_at, metric, value, geom)
        VALUES (%(dev_eui)s, %(received_at)s, %(metric)s, %(value)s,
                ST_GeomFromEWKT(%(geom_wkt)s))
        ON CONFLICT (dev_eui, received_at, metric) DO UPDATE
            SET value = EXCLUDED.value, geom = EXCLUDED.geom
        """,
        row,
    )


def transform(raw: dict) -> dict:
    """Normalize a raw sensor message into an output/sink row (domain-specific)."""
    return {
        "dev_eui": raw["dev_eui"],
        "received_at": raw["received_at"],
        "metric": raw["metric"],
        "value": float(raw["value"]),
        "geom_wkt": f"SRID=4326;POINT({raw['lon']} {raw['lat']})",
    }


def run_exactly_once(
    bootstrap: str,
    in_topic: str,
    out_topic: str,
    group_id: str,
    txn_id: str,
    pg_dsn: str,
    batch_size: int = 500,
    poll_timeout: float = 1.0,
) -> None:
    """Read-process-write loop with Kafka transactions and an idempotent PostGIS sink."""
    consumer, producer = build_clients(bootstrap, group_id, txn_id)
    consumer.subscribe([in_topic])
    producer.init_transactions()            # fence prior producer, once at startup
    pg = psycopg2.connect(pg_dsn)

    try:
        while True:
            batch = consumer.consume(num_messages=batch_size, timeout=poll_timeout)
            batch = [m for m in batch if m is not None and m.error() is None]
            if not batch:
                continue

            producer.begin_transaction()
            try:
                with pg:                     # one DB transaction per Kafka transaction
                    with pg.cursor() as cur:
                        for msg in batch:
                            row = transform(json.loads(msg.value()))
                            producer.produce(out_topic, json.dumps(row).encode())
                            upsert_reading(cur, row)

                # record consumed offsets INSIDE the Kafka transaction
                positions = [
                    TopicPartition(m.topic(), m.partition(), m.offset() + 1)
                    for m in batch
                ]
                producer.send_offsets_to_transaction(
                    positions, consumer.consumer_group_metadata()
                )
                producer.commit_transaction()
            except Exception:
                producer.abort_transaction()
                pg.rollback()
                raise
    finally:
        consumer.close()
        pg.close()

The DB commit and the Kafka commit are two separate systems, so a crash between them is still possible — that residual window is exactly why upsert_reading uses ON CONFLICT ... DO UPDATE. If the process dies after the PostGIS commit but before commit_transaction, the batch replays, the upsert is a no-op on the already-present rows, and no duplicate lands.

The ordering inside the loop is not arbitrary. The PostGIS work happens first, wrapped in its own with pg: block that commits the database transaction, and only then are offsets recorded and the Kafka transaction committed. This ordering means the residual failure window always resolves in the safe direction: if the crash lands between the two commits, the data is already durably in PostGIS and the replay merely re-confirms it via the upsert. Reversing the order — committing Kafka offsets first — would open a window where offsets advance but the rows were never written, and that is data loss, which no idempotency can recover. When forced to choose which system commits first, always let the idempotent one go first so a replay is a no-op rather than a gap.

Batch size is the main throughput lever and it trades against two costs. A larger batch amortizes the fixed cost of begin_transaction/commit_transaction — each transaction commit is a round trip to the transaction coordinator — across more messages, so raising the batch from 100 to 1000 can materially lift sustained throughput. But a larger batch holds the PostGIS transaction open longer, lengthening lock duration and the replay window on abort, and it delays offset commits so a rebalance sees more uncommitted work. For steady environmental telemetry arriving at a few hundred readings per second, batches of 250 to 1000 keep the transaction rate low enough to avoid coordinator pressure while keeping DB locks short.

Configuration Reference

The transactional guarantee depends on getting a handful of settings exactly right; defaults are not safe here.

Setting Client Value Why
transactional.id producer stable, unique per instance Enables zombie fencing across restarts
enable.idempotence producer True Required for transactions; dedupes producer retries
enable.auto.commit consumer False Offsets must flow through the transaction only
isolation.level consumer read_committed Downstream readers skip aborted-transaction messages
transaction.timeout.ms producer 60000 Must be ≤ broker transaction.max.timeout.ms
max.poll.interval.ms consumer ≥ batch processing time Prevents rebalance mid-transaction

Batch sizing: larger batches amortize the fixed transaction-commit cost but widen the replay window on abort and hold a DB transaction open longer. For steady sensor telemetry, 250–1000 messages per transaction balances throughput against lock duration. Timeout coupling: transaction.timeout.ms above the broker maximum makes init_transactions fail at startup — keep them in sync.

Verification and Testing

The natural-key upsert is what makes replay safe, so test it directly: applying the same row twice must leave exactly one record with the second row’s values.

import pytest


def test_upsert_is_idempotent(pg_cursor):
    row = {
        "dev_eui": "70B3D57ED0001A2B",
        "received_at": "2026-07-10T08:15:30+00:00",
        "metric": "temperature",
        "value": 24.6,
        "geom_wkt": "SRID=4326;POINT(4.9 52.37)",
    }

    upsert_reading(pg_cursor, row)
    upsert_reading(pg_cursor, {**row, "value": 24.6})   # exact replay after crash

    pg_cursor.execute(
        "SELECT count(*), max(value) FROM sensor_readings "
        "WHERE dev_eui = %s AND received_at = %s AND metric = %s",
        (row["dev_eui"], row["received_at"], row["metric"]),
    )
    count, value = pg_cursor.fetchone()
    assert count == 1        # replay did not duplicate
    assert value == 24.6


def test_abort_leaves_no_output(fake_producer):
    fake_producer.begin_transaction()
    fake_producer.produce("out", b'{"dev_eui": "x"}')
    fake_producer.abort_transaction()
    # a read_committed consumer must never observe the aborted message
    assert fake_producer.committed_records("out") == []

For an end-to-end check, run the loop against a test cluster, kill -9 the process mid-batch, restart it, and confirm the output topic (read with isolation.level=read_committed) and the PostGIS table each hold every reading exactly once. The requirement for a unique constraint on (dev_eui, received_at, metric) is what the ON CONFLICT clause depends on — create it before deploying.

Gotchas

Committing offsets through the consumer. Calling consumer.commit() alongside transactions breaks the atomicity — offsets then advance independently of the transaction. Disable auto-commit and route offsets exclusively through send_offsets_to_transaction.

Unstable transactional.id. Generating a fresh id per restart (for example from a UUID) disables fencing, so a hung previous instance can still commit and duplicate. Derive the id deterministically from the consumer group and partition assignment.

Forgetting read_committed downstream. The transaction is only half the protection; a downstream consumer left at the default read_uncommitted reads messages from aborted transactions and reintroduces duplicates.

Non-idempotent sink. Kafka transactions do not extend into PostGIS. Without a unique constraint and ON CONFLICT, the unavoidable crash window between the DB commit and the Kafka commit double-inserts on replay. The upsert is not optional.