Running Geometry Checks in a Kafka Consumer

A validating consumer sits between a raw event topic and everything downstream, and its job is narrow: decide, for each event, whether it is fit to pass. Doing that well is mostly about restraint — the checks themselves are microseconds of Shapely, while the ways to make a consumer slow, lossy or wrong are all in the plumbing around them. This guide builds the consumer: manual offset handling so nothing is lost, cost-ordered per-event checks, a cached reference index instead of per-event lookups, and dead-letter routing that produces a reprocessable message. It implements the per-event stage described in Streaming Spatial Data Validation.

Prerequisites

• Python 3.10+, confluent-kafka 2.3+, shapely 2.0+ and geopandas 0.14+ (for building the reference index at startup).
• A Kafka cluster with three topics: the input topic, a validated output topic and a dead-letter topic. Give the dead-letter topic a deliberate retention period; it will contain raw payloads.
• An event schema carrying an event identifier, a timestamp, a CRS code and a geometry in WKB, WKT or GeoJSON. Registered schemas with a schema registry are preferable; the example decodes JSON for readability.
• A reference layer small enough to hold in memory — operating areas, jurisdiction boundaries, permitted zones. If it is not small enough, the join belongs in a batch stage rather than in the consumer.
• A metrics sink for consumer lag, throughput and rejection rate. Running a validating consumer without lag metrics is the single most common way to discover a problem hours late.

Step-by-Step Procedure

Step 1 — Configure the consumer for correctness, not convenience

# consumer/step1_config.py
from confluent_kafka import Consumer, Producer

CONSUMER_CONF = {
    "bootstrap.servers": "kafka:9092",
    "group.id": "spatial-validator-v3",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,        # we commit after the outcome is durable
    "max.poll.interval.ms": 300_000,    # room for a slow reference refresh
    "session.timeout.ms": 45_000,
    "fetch.min.bytes": 65_536,          # batch a little; latency budget permitting
    "partition.assignment.strategy": "cooperative-sticky",
}

PRODUCER_CONF = {
    "bootstrap.servers": "kafka:9092",
    "enable.idempotence": True,
    "acks": "all",
    "linger.ms": 20,
    "compression.type": "zstd",
}


def build():
    return Consumer(CONSUMER_CONF), Producer(PRODUCER_CONF)

Verification: confirm enable.auto.commit is false in the running configuration, not just in the file. The single most common cause of “we lost events during a deploy” is an auto-commit left enabled in an environment override.

Step 2 — Decode and screen before constructing geometry

# consumer/step2_decode.py
import json

MAX_ABS_LON, MAX_ABS_LAT = 180.0, 90.0
CANONICAL_EPSG = 4326


def decode(raw: bytes) -> dict:
    """Cheapest checks first: parse, required fields, coordinate sanity, CRS."""
    try:
        event = json.loads(raw)
    except json.JSONDecodeError as exc:
        return {"ok": False, "rule": "STR_PARSE_001", "reason": f"invalid JSON: {exc}"}

    for field in ("event_id", "ts", "crs", "geometry"):
        if field not in event:
            return {"ok": False, "rule": "STR_SCHEMA_001",
                    "reason": f"missing required field {field!r}"}

    if int(event["crs"]) != CANONICAL_EPSG:
        return {"ok": False, "rule": "STR_CRS_001",
                "reason": f"crs {event['crs']} != canonical EPSG:{CANONICAL_EPSG}"}

    coords = event["geometry"].get("coordinates")
    if event["geometry"].get("type") == "Point":
        lon, lat = coords[0], coords[1]
        if abs(lon) > MAX_ABS_LON or abs(lat) > MAX_ABS_LAT:
            return {"ok": False, "rule": "STR_RANGE_001",
                    "reason": f"coordinates out of range: {lon}, {lat}"}

    return {"ok": True, "event": event}

Verification: feed the decoder a truncated payload, a payload with a projected coordinate pair, and a valid one. Each must return a distinct rule identifier — the reason a consumer is worth building is that it distinguishes these, where a generic try/except would report them all as “bad event”.

Step 3 — Run the geometry rules in cost order

# consumer/step3_geometry.py
import shapely
from shapely.geometry import shape


def check_geometry(event: dict) -> dict:
    try:
        geom = shape(event["geometry"])
    except Exception as exc:                      # malformed coordinate structure
        return {"ok": False, "rule": "STR_GEOM_001",
                "reason": f"unreadable geometry: {exc}"}

    if geom.is_empty:
        return {"ok": False, "rule": "STR_GEOM_002", "reason": "empty geometry"}

    if not shapely.is_valid(geom):
        return {"ok": False, "rule": "STR_GEOM_003",
                "reason": shapely.is_valid_reason(geom)}

    return {"ok": True, "geom": geom}

Verification: time this function over ten thousand real events. A point stream should average well under fifty microseconds per event; if it does not, the payloads are larger than expected — polygons with thousands of vertices arriving on what was assumed to be a point stream, which is itself worth a finding.

Step 4 — Join against a cached, refreshable reference index

# consumer/step4_reference.py
import threading
import time

import geopandas as gpd
from shapely import STRtree


class ReferenceIndex:
    """An in-memory spatial index, swapped atomically on refresh."""

    def __init__(self, path: str, refresh_seconds: int = 900):
        self._path = path
        self._lock = threading.Lock()
        self._tree, self._frame = self._load()
        threading.Thread(target=self._refresh_loop, args=(refresh_seconds,),
                         daemon=True).start()

    def _load(self):
        frame = gpd.read_file(self._path).to_crs(4326)
        return STRtree(frame.geometry.values), frame

    def _refresh_loop(self, seconds: int):
        while True:
            time.sleep(seconds)
            try:
                tree, frame = self._load()
                with self._lock:                # atomic swap; readers never see a partial index
                    self._tree, self._frame = tree, frame
            except Exception:
                pass                            # keep serving the previous index

    def containing(self, geom):
        with self._lock:
            tree, frame = self._tree, self._frame
        for idx in tree.query(geom):
            if frame.geometry.iloc[idx].contains(geom):
                return frame.iloc[idx]
        return None

Verification: measure the lookup at steady state — a query against an STRtree of a few thousand polygons is a handful of microseconds. Then compare against the naive alternative of a SELECT ... ST_Contains per event; the difference is typically three orders of magnitude, and it is the difference between a consumer that keeps up and one that accumulates lag forever.

Step 5 — Produce the outcome, then commit

# consumer/step5_loop.py
import json

VALID_TOPIC, DLQ_TOPIC, CONSUMER_VERSION = "spatial.validated", "spatial.dlq", "v3.1.0"


def dead_letter(producer, msg, rule: str, reason: str) -> None:
    producer.produce(
        DLQ_TOPIC,
        key=msg.key(),
        value=msg.value(),                       # original bytes, unmodified
        headers=[
            ("rule", rule.encode()),
            ("reason", reason[:400].encode()),
            ("consumer_version", CONSUMER_VERSION.encode()),
            ("source_topic", msg.topic().encode()),
            ("source_partition", str(msg.partition()).encode()),
            ("source_offset", str(msg.offset()).encode()),
        ],
    )


def run(consumer, producer, reference, decode, check_geometry, metrics):
    consumer.subscribe(["spatial.raw"])
    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                continue
            if msg.error():
                metrics.increment("consume_error")
                continue

            decoded = decode(msg.value())
            if not decoded["ok"]:
                dead_letter(producer, msg, decoded["rule"], decoded["reason"])
            else:
                event = decoded["event"]
                geom_result = check_geometry(event)
                if not geom_result["ok"]:
                    dead_letter(producer, msg, geom_result["rule"], geom_result["reason"])
                else:
                    area = reference.containing(geom_result["geom"])
                    if area is None:
                        dead_letter(producer, msg, "STR_AOI_001",
                                    "event falls outside every operating area")
                    else:
                        event["area_id"] = str(area["area_id"])
                        event["validated_by"] = CONSUMER_VERSION
                        producer.produce(VALID_TOPIC, key=msg.key(),
                                         value=json.dumps(event).encode())

            producer.flush()          # outcome is durable...
            consumer.commit(msg)      # ...only now is the offset safe to advance
            metrics.increment("processed")
    finally:
        producer.flush()
        consumer.close()

Verification: kill the consumer between the produce and the commit — with a debugger breakpoint or a SIGKILL under load — and restart it. The event must be reprocessed and produce an identical outcome. That is at-least-once delivery working as intended, and it is only safe because the validation is deterministic.

Interpreting Results

The consumer’s own metrics matter as much as its findings:

STR_AOI_001 deserves attention because it is the rule most often wrong in the validator rather than in the data. Events falling outside every operating area usually mean the reference layer is stale — a new depot, a boundary change — rather than that the events are bad. Alerting on a sudden rise in that rule specifically has caught more reference-data problems than data problems in practice.

Gotchas & Edge Cases

Committing before producing loses events. It looks harmless because the happy path never exercises the gap. Order the operations produce-then-commit and flush in between, or accept that a rebalance during a deploy will drop whatever was in flight.

producer.flush() per message is slow, and per batch is complicated. The example flushes per message for clarity. In production, accumulate a batch of messages and offsets, flush once, then commit the highest offset — throughput improves by an order of magnitude and the correctness argument is unchanged as long as the commit follows the flush.

A rebalance during a long reference refresh looks like a crash. If the refresh blocks the poll loop past max.poll.interval.ms, Kafka evicts the consumer. Running the refresh on a background thread, as in Step 4, avoids this; running it inline does not.

Dead-letter loops are easy to create. A repair consumer that reads the dead-letter topic and republishes to the input topic without fixing anything creates an infinite circulation. Always stamp a retry count in the headers and stop after a bounded number of passes.

Schema evolution breaks decoders silently. A new optional field is harmless; a renamed field turns every event into STR_SCHEMA_001. Use a schema registry with compatibility rules if the producer is not under your control, and alert on a rejection-rate step change rather than on an absolute threshold.

Ordering guarantees are per partition, not per topic. Sequence rules only work if all events for an entity share a partition, which means keying by entity identifier. Keying by event identifier — a natural-looking choice — scatters an entity’s history across partitions and quietly breaks every sequence check.

When to Escalate

• Sustained lag after profiling shows the per-event path is fast — the bottleneck is partition count or consumer instances, which is a capacity conversation rather than a code change.
• A rejection-rate step change with no deploy points upstream: a producer changed its payload, a device fleet was updated, or a schema evolved. Escalate with the rule breakdown, which usually names the change precisely.
• Dead-letter volume that nobody consumes is a governance problem. Assign an owner to the topic under the model in Assigning Spatial Data Ownership with a RACI Matrix, or stop producing to it.
• Rules that need neighbours or history do not belong in this consumer. Move them to the windowed stage described in Handling Late and Out-of-Order Spatial Events rather than adding state here.

Frequently Asked Questions

Related

• Streaming Spatial Data Validation — rule classification and the wider streaming design
• Validating GPS Tracks for Speed and Teleport Outliers — the stateful rules that need per-entity keying
• Designing Async Validation Queues with Celery — the task-queue alternative when latency is measured in minutes

Back to Streaming Spatial Data Validation