Exactly-Once Semantics for Streaming Trajectory Matching

The windowed consumer in the previous page gives at-least-once delivery: on a crash between emit and offset-commit, fixes replay and matched edges are re-produced, deduplicated only by an application-level emitted_upto guard. That guard is fragile — it lives in volatile per-vehicle state that a rebalance can reset. This page removes the fragility by making the emit and the offset advance a single atomic Kafka transaction, so each matched segment on the output topic appears exactly once no matter how many times the input is retried or which consumer instance ends up owning the partition. It is the delivery-guarantee layer of the broader streaming map-matching pipeline.

The specific hazard exactly-once addresses in trajectory matching is duplicate mileage. If the same matched edge is emitted twice, downstream distance, toll, and billing aggregations over-count, and the error is invisible until a reconciliation months later. A transaction binds the matched-edge output to the read position atomically: either both advance or neither does.

Compatibility and Configuration Requirements

Tables scroll horizontally on narrow viewports.

Transactional Consume-Process-Produce

The example below wraps one poll batch in a Kafka transaction. It produces matched edges and stages the consumed offsets inside the same transaction, then commits both atomically. On any failure it aborts, and the batch replays with no partial output. This is the exactly-once counterpart of the plain consumer’s produce-flush-commit sequence.

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


class ExactlyOnceMatcher:
    """
    Transactional streaming map-matcher: each matched edge is emitted exactly
    once across the consume-process-produce cycle.

    Parameters
    ----------
    brokers : str
        Kafka bootstrap servers.
    group_id : str
        Consumer group; splits input partitions across instances.
    transactional_id : str
        STABLE id that outlives restarts. Fences zombie producers after a
        rebalance so a stale instance cannot commit a duplicate segment.
    in_topic / out_topic : str
        Source fixes topic and destination matched-edges topic.
    matcher : object
        Windowed decoder exposing match_window(fixes, lag) -> [edge dicts].
    """

    def __init__(self, brokers, group_id, transactional_id,
                 in_topic, out_topic, matcher, lag=8):
        self.in_topic = in_topic
        self.out_topic = out_topic
        self.matcher = matcher
        self.lag = lag
        self.group_id = group_id
        self.windows = {}  # vehicle_id -> list[fix]; see note on state below

        self.consumer = Consumer({
            "bootstrap.servers": brokers,
            "group.id": group_id,
            "enable.auto.commit": False,        # offsets ride inside the txn
            "auto.offset.reset": "latest",
            # read_committed so we never process another stage's aborted output
            "isolation.level": "read_committed",
        })
        self.producer = Producer({
            "bootstrap.servers": brokers,
            "transactional.id": transactional_id,   # enables idempotence + txns
            "enable.idempotence": True,
            "transaction.timeout.ms": 60000,        # must exceed batch time
        })
        # init_transactions fences any older producer with the same id (zombie)
        self.producer.init_transactions()

    def _positions(self, batch):
        """Highest consumed offset per partition, +1 = next to read."""
        pos = {}
        for msg in batch:
            key = (msg.topic(), msg.partition())
            pos[key] = max(pos.get(key, -1), msg.offset())
        return [TopicPartition(t, p, off + 1) for (t, p), off in pos.items()]

    def run(self):
        self.consumer.subscribe([self.in_topic])
        try:
            while True:
                batch = self.consumer.consume(num_messages=500, timeout=1.0)
                batch = [m for m in batch if m and not m.error()]
                if not batch:
                    continue

                self.producer.begin_transaction()
                try:
                    for msg in batch:
                        vehicle_id = msg.key().decode("utf-8")
                        fix = json.loads(msg.value())
                        for edge in self._match(vehicle_id, fix):
                            self.producer.produce(
                                self.out_topic,
                                key=vehicle_id.encode("utf-8"),
                                value=json.dumps(edge).encode("utf-8"))

                    # stage consumed offsets INSIDE the transaction so the read
                    # position advances atomically with the matched-edge output
                    self.producer.send_offsets_to_transaction(
                        self._positions(batch),
                        self.consumer.consumer_group_metadata())
                    self.producer.commit_transaction()
                except Exception:
                    # abort: neither edges nor offsets take effect; batch replays
                    self.producer.abort_transaction()
                    raise
        finally:
            self.consumer.close()

    def _match(self, vehicle_id: str, fix: dict) -> list:
        """Append to the per-vehicle window and return committed edges."""
        window = self.windows.setdefault(vehicle_id, [])
        window.append(fix)
        if len(window) > 60:               # bound the window
            window.pop(0)
        if len(window) <= self.lag:
            return []                      # not enough context to commit
        return self.matcher.match_window(list(window), lag=self.lag)

Execution and Tuning Guidelines

Run one instance per partition group, each with its own stable transactional_id. Downstream consumers of out_topic — dashboards, mileage aggregators, billing — must set isolation.level=read_committed; otherwise they read aborted records and the guarantee is lost at the last hop.

• transactional.id — the fencing key. It must be stable across restarts and unique per logical producer. When an instance restarts, init_transactions() reuses the id, bumps the producer epoch, and fences the old (zombie) instance. A random id per boot silently disables fencing and re-admits duplicates. Derive it deterministically from the assigned partition group, never from a UUID or hostname that changes.
• isolation.level=read_committed — set on every consumer that reads matched edges, including this matcher’s own input if an upstream stage is transactional. It makes aborted and in-flight transactional records invisible. Leaving it at the default read_uncommitted means a downstream reader sees edges from transactions that later abort — the exact duplicate you were preventing.
• enable.idempotence=true — implied once transactional.id is set, but state it explicitly. It dedupes producer retries within a session via sequence numbers; transactions extend that to atomicity across the consume-process-produce cycle. Idempotence alone is not exactly-once for the pipeline.
• transaction.timeout.ms — must exceed the worst-case time to process one batch (window decode plus any routing-engine calls) and stay under the broker’s transaction.max.timeout.ms. Too low and the coordinator aborts a healthy transaction under load, forcing needless replays; too high and a genuinely stuck instance holds the partition’s LSO (last stable offset) and stalls downstream read_committed readers.
• Batch size — larger transactions amortise the commit round-trip but widen the replay window on abort and delay the LSO advance, adding downstream latency. Tune against the emission latency budget from the pipeline page.

Exactly-once covers only what lives in Kafka: the matched-edge output and the consumed offsets. It does not cover per-vehicle window state held outside Kafka. Rebuild that state from a changelog on assignment (as the windowed consumer does) or accept that windows rewarm after a restart while the delivery guarantee still holds. Feeding the matcher clean, Kalman-filtered fixes keeps decode time — and therefore transaction.timeout.ms headroom — predictable.

Common Pitfalls

Up: Streaming Map-Matching Pipelines | Trajectory Analysis & Map Matching Techniques

Related

• Streaming Map-Matching Pipelines — the end-to-end design this delivery guarantee slots into
• Kafka Consumer Windowed Map Matching in Python — the at-least-once consumer this page upgrades to transactional exactly-once
• Hidden Markov Model Map Matching in Python — the decoder producing the matched edges the transaction commits
• OSRM Integration for Fleet Map Matching — the routing-engine calls whose latency sets the transaction timeout budget
• Kalman Filtering for GPS Noise Reduction — upstream smoothing that keeps per-batch decode time predictable