Deduplicating EPCIS Events in a Kafka Stream

Kafka’s at-least-once delivery guarantee means every ObjectEvent your packaging line commissions can reach your consumer more than once — after a broker retry, a producer retry following an ack timeout, or a consumer-group rebalance that replays from the last committed offset. This sits within Real-Time Event Stream Processing, the area covering the message-broker layer beneath Serialization Data Ingestion & EPCIS Event Sync, and it becomes a compliance problem the instant a redelivered commissioning event double-counts a serial or a duplicate decommission retires the same unit twice. The fix is not “exactly-once” Kafka semantics — those only guarantee exactly-once within the broker/producer transaction, never against your downstream EPCIS repository — it is an idempotent consumer that recognizes true redeliveries while still accepting every legitimate re-scan of the same GTIN (01) and serial (21). GS1’s EPCIS 2.0 specification recommends every event carry an eventID, the strongest signal a dedup filter can use; this guide covers what to do when that field is missing or inconsistent across trading partners.

Prerequisites #

• Python 3.10+ — the snippets use X | Y-style typing and tz-aware datetime throughout.
• aiokafka ≥ 0.10 — an asyncio-native Kafka client with manual offset commit and rebalance-listener hooks.
• redis ≥ 5.0 (redis.asyncio) — for the seen-set. A high-throughput line can substitute an embedded RocksDB state store keyed the same way, but Redis’s SET ... NX EX gives an atomic check-and-set with a TTL in one round trip, which keeps the filter safe under concurrent consumers.
• fakeredis (dev/test only) — an in-memory Redis stand-in for the pytest scaffold in Verification.
• DSCSA data prerequisites — inbound ObjectEvent records carrying the SGTIN pool (GTIN (01) + serial (21), ideally alongside lot (10) and expiry (17)), a bizStep, and a tz-aware eventTime. An eventID is not guaranteed by every trading partner, so the filter must work with or without one.

Before writing code, confirm how your producer partitions the topic. If it partitions by (gtin, serial), redeliveries for one unit always land on the same consumer instance during normal operation, and only a rebalance moves that responsibility to a different process — which is exactly the scenario this guide’s dedup store has to survive.

Step-by-Step Solution #

Step 1 — Derive a deterministic idempotency key #

Two independent deliveries of the same physical scan must always hash to the same key; two independent scans must never collide. Prefer the event’s own eventID when present, and fall back to a hash of the fields that together identify one scan: GTIN (01), serial (21), bizStep, and eventTime.

import hashlib
from datetime import datetime, timezone
from typing import Optional


def build_idempotency_key(event: dict) -> str:
    """Derive a stable idempotency key for a possibly-redelivered ObjectEvent.

    Prefers the event's own eventID (EPCIS 2.0 events SHOULD carry one).
    Falls back to a deterministic hash of the fields that identify a
    single physical scan: GTIN (01), serial (21), bizStep, and eventTime.
    """
    event_id: Optional[str] = event.get("eventID")
    if event_id:
        return f"evt:{event_id}"

    gtin: str = event["epc_gtin"]        # AI (01), keep as zero-padded str
    serial: str = event["epc_serial"]    # AI (21)
    biz_step: str = event["bizStep"]
    event_time: datetime = event["eventTime"]  # must be tz-aware

    canonical = "|".join([
        gtin,
        serial,
        biz_step,
        event_time.astimezone(timezone.utc).isoformat(),
    ])
    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
    return f"hash:{digest}"

DSCSA/GS1 note: the hash fallback must include bizStep — omitting it would collapse a commissioning ObjectEvent and a later shipping ObjectEvent for the same GTIN (01)/serial (21) into one key, silently discarding the second as a false duplicate instead of recording a distinct, legally required lifecycle transition.

Step 2 — A Redis-backed seen-set with atomic SETNX and TTL #

SET key value NX EX ttl is a single atomic Redis command: there is no gap between “check if seen” and “mark as seen” for a second consumer to race into, which matters the moment a rebalance hands the same partition to another process mid-flight.

import redis.asyncio as redis


class RedisDedupStore:
    """Atomic 'have I seen this key before' check backed by Redis."""

    def __init__(self, client: redis.Redis, ttl_seconds: int = 172_800):
        self._client = client
        self._ttl = ttl_seconds  # 48h default: longer than any realistic replay lag

    async def is_new(self, idempotency_key: str) -> bool:
        """Return True the first time a key is seen, False on every redelivery."""
        was_set = await self._client.set(
            name=f"epcis:dedup:{idempotency_key}",
            value="1",
            nx=True,
            ex=self._ttl,
        )
        return bool(was_set)

DSCSA/GS1 note: the TTL bounds how long the seen-set retains a key — pick it longer than the worst-case Kafka redelivery window (broker retry plus rebalance replay), but it is not a substitute for the durable EPCIS repository’s six-year DSCSA retention; the seen-set is a transient dedup cache, not the record of truth.

Step 3 — Wire the filter into an async consumer loop #

Disable Kafka’s auto-commit so the offset only advances after the dedup check and, when the event is new, after the downstream write succeeds. A duplicate still needs its offset committed, or the partition stalls forever behind the same message.

import json
from aiokafka import AIOKafkaConsumer


async def consume_and_dedup(
    topic: str,
    bootstrap_servers: str,
    group_id: str,
    dedup_store: RedisDedupStore,
    epcis_repo,
) -> None:
    consumer = AIOKafkaConsumer(
        bootstrap_servers=bootstrap_servers,
        group_id=group_id,
        enable_auto_commit=False,   # commit only after dedup + processing succeed
        auto_offset_reset="earliest",
    )
    consumer.subscribe([topic])
    await consumer.start()
    try:
        async for msg in consumer:
            event = json.loads(msg.value)
            key = build_idempotency_key(event)

            if not await dedup_store.is_new(key):
                # True redelivery: this key was already marked seen. Skip the
                # side effect but still advance the offset, otherwise this
                # partition never progresses past the duplicate.
                await consumer.commit()
                continue

            await epcis_repo.commit_object_event(event)
            await consumer.commit()
    finally:
        await consumer.stop()

DSCSA/GS1 note: committing the offset on both branches keeps the consumer’s position monotonic; only the ObjectEvent write to the repository is conditional on the key being new, so a duplicate never touches the audit trail twice.

Step 4 — Handle redelivery after a consumer-group rebalance #

A rebalance is what breaks naive, in-memory deduplication. When a partition moves to a new consumer process, the new owner resumes from the last committed offset — which can be at or before the last message the previous owner actually processed — so it may redeliver an event already written to the repository moments before the handoff. An in-process set() would have no memory of that write; the Redis-backed store does, because it lives outside any single consumer.

from aiokafka import ConsumerRebalanceListener


class DedupAwareRebalanceListener(ConsumerRebalanceListener):
    """Logs rebalances; the dedup store needs no special handling here
    because it lives in Redis, not in consumer-local memory."""

    def __init__(self, logger):
        self._log = logger

    async def on_partitions_revoked(self, revoked):
        self._log.info("partitions revoked: %s", revoked)

    async def on_partitions_assigned(self, assigned):
        self._log.info(
            "partitions assigned: %s — resuming from last committed offset",
            assigned,
        )

Attach it with consumer.subscribe([topic], listener=DedupAwareRebalanceListener(logger)) in place of the plain subscribe([topic]) call in Step 3. No dedup-specific logic belongs in the listener itself — its only job is observability, because the correctness guarantee already lives in the external seen-set that both the old and new partition owner can see.

Step 5 — Distinguish a true duplicate from a legitimate re-scan #

Dropping redeliveries must never mean dropping a genuinely new ObjectEvent for the same unit. A unit can legitimately be observed twice — once at commissioning, again at a shipping read point hours later — and both are distinct events the repository must retain. The key handles this correctly because it includes bizStep and eventTime: a true redelivery carries identical values for both, while a legitimate re-scan changes at least one, producing a different key that passes straight through.

from datetime import datetime, timezone

base_scan = {
    "epc_gtin": "00312345000019",
    "epc_serial": "SERIAL001",
    "bizStep": "urn:epcglobal:cbv:bizstep:shipping",
    "eventTime": datetime(2026, 7, 17, 10, 0, 0, tzinfo=timezone.utc),
}
replayed = dict(base_scan)  # identical redelivery of the same physical scan
rescanned = dict(base_scan, eventTime=datetime(2026, 7, 17, 14, 30, 0, tzinfo=timezone.utc))

assert build_idempotency_key(base_scan) == build_idempotency_key(replayed)
assert build_idempotency_key(base_scan) != build_idempotency_key(rescanned)

DSCSA/GS1 note: if a trading partner’s clocks only resolve to whole seconds and a read point can genuinely fire twice within one second, widen the key with a partner-supplied sequence number or the readPoint GLN rather than shortening the TTL — narrowing the dedup window does nothing to fix a key that is under-specified.

Verification #

Confirm the filter drops redeliveries but never a distinct scan, using an in-memory Redis stand-in so the test suite has no external dependency.

import pytest
import fakeredis.aioredis


@pytest.mark.asyncio
async def test_seen_set_drops_redelivery():
    client = fakeredis.aioredis.FakeRedis()
    store = RedisDedupStore(client, ttl_seconds=60)

    key = "hash:abc123"
    assert await store.is_new(key) is True    # first delivery
    assert await store.is_new(key) is False   # redelivered — dropped
    assert await store.is_new(key) is False   # rebalance replay — still dropped


def test_distinguishes_replay_from_rescan():
    base = {
        "epc_gtin": "00312345000019",
        "epc_serial": "SERIAL001",
        "bizStep": "urn:epcglobal:cbv:bizstep:shipping",
        "eventTime": "2026-07-17T10:00:00+00:00",
    }
    assert build_idempotency_key({**base, "eventID": "e-1"}) == "evt:e-1"
    assert build_idempotency_key({**base, "eventID": "e-1"}) != build_idempotency_key(
        {**base, "eventID": "e-2"}
    )

Run both under pytest -k dedup. In production, watch the Redis key count for epcis:dedup:* — it should track with unique-event throughput times the TTL window, not grow unbounded — and check consumer-group lag after a forced rebalance to confirm the partition resumes without a spike in duplicate repository writes.

Gotchas & Edge Cases #

• Leading-zero GTINs. Always keep the GTIN (01) as a string in the key material. Casting it to int strips a leading zero, changing the hash on any path that re-derives the key from re-parsed data and breaking dedup silently across a backfill.
• UTC vs. local eventTime. Normalize to UTC before hashing. Identical redeliveries parsed with different local-timezone assumptions produce different canonical strings and keys, defeating the filter without raising any error.
• Under-specified hash collisions. If the fallback hash omits bizStep or eventTime, two genuinely different EPCIS events for the same unit collapse onto one key and the second is dropped as a false duplicate — harder to notice because nothing errors.
• TTL shorter than the real redelivery window. A consumer stuck behind a slow rebalance or a broker-side retry can replay a message after a short TTL has expired, so the seen-set no longer recognizes it and a true duplicate slips through as if new.
• Marking seen before the write is durable. If the Redis key is set before the ObjectEvent write, a crash between the two loses the event permanently — marked seen but never persisted. Mark seen only after the write succeeds, or make the repository write itself idempotent as defense in depth.

Related #

• Up to the parent section: Real-Time Event Stream Processing
• Kafka vs. RabbitMQ for Serialization Event Streaming — choosing the broker this dedup pattern runs against
• Idempotent Batch Reprocessing with Python — the same idempotency-key discipline applied to batch replays instead of a live stream
• Serialization Data Ingestion & EPCIS Event Sync — the pipeline this stream-processing stage feeds