Consumer-Group Rebalancing Without Dropping Geofence Triggers

Every time a geofence consumer group scales up, loses a pod, or restarts on deploy, Kafka rebalances — it revokes partitions from some members and reassigns them to others. Done naively, that rebalance is exactly when triggers go missing: a consumer that committed offsets it had not yet processed hands a partition to a peer that resumes past the unprocessed ENTER, and the vehicle that crossed the fence never gets billed. This page sits under Kafka vs Redis Streams for geofence trigger routing and the broader event routing and backpressure layer, and it answers the operational question that surfaces on the first production deploy: how do you configure and code a consumer group so a routine rebalance costs zero dropped and zero double-processed triggers? The answer is four settings and one commit discipline — the cooperative-sticky assignor, static membership, a tuned max.poll.interval.ms, and manual offset commit strictly after the side effect — backed by idempotent handling so any unavoidable replay is absorbed downstream.

Concept and specification #

A rebalance is the protocol by which a consumer group agrees on which member owns which partition. The legacy eager protocol is stop-the-world: on any membership change every consumer revokes all its partitions, the group leader recomputes assignments, and everyone resumes — so a single pod restart pauses the entire group and re-reads from the last committed offset. The cooperative-sticky protocol revokes only the partitions that actually move, letting untouched partitions keep processing throughout, which shrinks both the pause and the window in which drops can happen.

Two clocks bound a safe rebalance. A consumer must call poll at least every max.poll.interval.ms, or the broker assumes it is dead and evicts it — triggering a rebalance and reprocessing. And it must send a heartbeat within session.timeout.ms, or it is removed from the group. The trigger-drop hazard lives at the offset boundary: an offset commit tells the group “everything up to here is processed,” so committing before the side effect completes converts a rebalance into silent loss, while committing after — with idempotent handling — converts it into at-most-harmless replay.

The safety invariant is a strict ordering per partition:

$$\text{fetch}\to \text{process (idempotent)}\to \text{commit offset}$$

Reverse the last two and a crash or revoke between commit and process drops every trigger in the gap; keep them in order and the worst case is reprocessing a batch whose effects the idempotency key already absorbed.

Step-by-step implementation #

Prerequisites: Python 3.11+, aiokafka>=0.10, an idempotency store (Redis or a keyed table) so replayed triggers are absorbed. Input is the partitioned geofence.triggers topic; output is exactly-once-effective downstream state across rebalances.

1. Configure the consumer for cooperative, static membership with manual commit. The group_instance_id must be stable per pod (inject the StatefulSet ordinal or pod name), so a quick restart rejoins as the same member and keeps its partitions.

from aiokafka import AIOKafkaConsumer  # aiokafka >= 0.10

async def build_consumer(pod_id: str) -> AIOKafkaConsumer:
    consumer = AIOKafkaConsumer(
        "geofence.triggers",
        bootstrap_servers="broker-1:9092,broker-2:9092,broker-3:9092",
        group_id="trigger-router",
        group_instance_id=pod_id,          # static membership: survive quick restarts
        partition_assignment_strategy=[
            "cooperative-sticky",           # revoke only partitions that move
        ],
        enable_auto_commit=False,           # commit manually, after the side effect
        session_timeout_ms=45_000,          # ride out a rolling restart
        max_poll_interval_ms=300_000,       # above worst-case batch processing time
        max_poll_records=256,               # bound the batch so it finishes within the interval
    )
    await consumer.start()
    return consumer

Gotcha: group_instance_id must be unique per pod but stable across restarts. Two live pods sharing one id triggers a fencing error; a value that changes on every restart (a random UUID) defeats static membership entirely, forcing a reassignment on every deploy.

2. Process the batch, then commit — never the reverse. Handle every record, apply the idempotency key so a replay is a no-op, and only then commit the batch’s offsets.

async def run(consumer: AIOKafkaConsumer, seen: "IdempotencyStore") -> None:
    while True:
        batches = await consumer.getmany(timeout_ms=1_000, max_records=256)
        for tp, records in batches.items():
            for rec in records:
                key = f"{rec.key.decode()}:{rec.offset}"  # dedup key
                if await seen.check_and_set(key):
                    continue                                # already applied: replay no-op
                await _apply_trigger(rec.value)             # the side effect
            # Commit this partition's offsets ONLY after every record is processed.
            await consumer.commit({tp: records[-1].offset + 1})

Gotcha: commit the offset of the last processed record plus one, per partition, not a blanket commit() — a blanket commit after a partially processed batch re-introduces exactly the loss window this whole procedure exists to close.

3. Handle revocation to finish in-flight work before the partition moves. Register an on_partitions_revoked listener that commits completed offsets so the new owner resumes cleanly.

from aiokafka import ConsumerRebalanceListener

class CommitOnRevoke(ConsumerRebalanceListener):
    def __init__(self, consumer: AIOKafkaConsumer) -> None:
        self._c = consumer
    async def on_partitions_revoked(self, revoked) -> None:
        await self._c.commit()   # flush processed offsets before the partition moves
    async def on_partitions_assigned(self, assigned) -> None:
        pass                     # cooperative: only newly-added partitions arrive here

Gotcha: a poison message that never processes will hold a partition’s offset indefinitely and stall it through every rebalance. Cap retries and divert it to a dead-letter path — see dead-letter topics and poison-message handling — so one bad trigger cannot freeze a partition across the group.

Benchmark and verification #

The change to measure is triggers dropped and triggers double-processed per rebalance, before and after applying the configuration above, under a controlled rebalance: a rolling restart of one consumer in a 6-pod group draining a 12-partition topic at 40k triggers/sec. “Before” is auto-commit with the eager assignor; “after” is manual commit-after-process with cooperative-sticky and static membership.

async def measure_rebalance(consumer, sink_counter) -> dict[str, int]:
    # Compare the set of trigger keys committed as processed against the set
    # actually applied to the downstream sink across one rebalance window.
    committed = await _read_committed_offsets(consumer)
    applied = sink_counter.applied_keys()
    return {
        "dropped": len(committed - applied),      # committed but never applied == loss
        "duplicated": sink_counter.duplicate_applies(),  # absorbed by idempotency
    }

Representative figures across 20 controlled rebalances:

The drops collapse to zero because no offset is ever committed ahead of the side effect, so a revoke can only cause the new owner to reprocess — and those reprocesses show up in the duplicate row, absorbed by the idempotency key rather than corrupting downstream state. Cooperative-sticky is why only two partitions pause instead of twelve, which is why the P99 during rebalance drops from 410ms to 22ms. Verify the same way in staging: drive a rolling restart under load, diff the set of committed offsets against the set of side effects actually applied, and require the dropped count to be exactly zero before promoting the config. Confirm duplicates are all absorbed by checking the idempotency store’s hit counter rises during the rebalance window.

Failure modes and edge cases #

• Commit before process (the classic). Any path that commits an offset before the side effect completes — auto-commit on a timer, or a manual commit at the top of the loop — drops every trigger between the committed offset and the crash point. Keep commit strictly last, per partition.
• Long poll eviction. A batch that takes longer than max.poll.interval.ms to process is treated as a dead consumer: the broker evicts it, rebalances, and the batch is reprocessed elsewhere while this consumer discovers it was fenced only on its next commit. Bound max_poll_records so the worst-case batch finishes well inside the interval, or offload heavy geometry so the poll loop stays responsive.
• Poison message stalling a partition. A trigger that raises on every attempt never lets its offset advance, so the partition is stuck through every subsequent rebalance and its lag grows without bound. Cap retries and divert to the dead-letter path; never loop forever on one record.
• Duplicate applies without idempotency. Commit-after-process guarantees at-least-once, which will replay on rebalance. Without an idempotency key that replay double-charges a fare or double-fires a compliance alert. The key derived from (device_id, fence_id, transition, event_time_bucket) — detailed in idempotent trigger emission semantics — is not optional here; it is what makes the safe commit order safe.
• Unstable group.instance.id. A per-restart random id defeats static membership, so every deploy pays a full reassignment; two pods sharing an id fence each other. Bind the id to a stable pod identity.
• Rolling deploy thundering herd. Restarting all pods at once still forces churn even with cooperative-sticky. Stagger the rollout so at most one or two members are in flux at a time, keeping the moved-partition set — and the replay volume — small.

Related #

• Kafka vs Redis Streams for Geofence Trigger Routing — parent comparison of broker substrates and their consumer-group semantics.
• Tuning Kafka Producer Batching for Geofence Triggers — the produce-side counterpart to safe consumption.
• Dead-Letter Topics and Poison-Message Handling — diverting the poison message that would otherwise stall a partition across rebalances.