Problem 97 · Data Engineering & Streaming

Stream Deduplication with Idempotency Keys and Watermark

HardClassic DSA & systems

01Problem

Process an at-least-once event stream so each idempotency key is emitted exactly once, keeping bounded state by evicting keys older than a watermark and dropping events that arrive too late.

02Key insight

At-least-once delivery + idempotent dedupe = effectively-once. Unbounded seen-sets blow up, so bound state with event-time watermark = max_event_time - allowed_lateness: keys older than the watermark are evicted, and any event older than the watermark is rejected as late (a duplicate can't sneak in after eviction).

03Solution

solution.pypython31 lines

import heapq
from typing import Hashable


class StreamDeduplicator:
    def __init__(self, allowed_lateness: float) -> None:
        self.allowed_lateness = allowed_lateness
        self.watermark = float("-inf")
        self._seen: dict[Hashable, float] = {}
        self._expiry: list[tuple[float, Hashable]] = []  # min-heap by event time
        self.duplicates = 0
        self.late = 0

    def process(self, key: Hashable, event_time: float) -> bool:
        """Return True iff the event should be emitted downstream."""
        if event_time < self.watermark:
            self.late += 1  # in production: route to a dead-letter/side output
            return False
        if key in self._seen:
            self.duplicates += 1
            return False
        self._seen[key] = event_time
        heapq.heappush(self._expiry, (event_time, key))
        self.watermark = max(self.watermark, event_time - self.allowed_lateness)
        while self._expiry and self._expiry[0][0] < self.watermark:
            _, old = heapq.heappop(self._expiry)
            del self._seen[old]  # safe: a retry of `old` would now be rejected as late
        return True

    def state_size(self) -> int:
        return len(self._seen)

Try it first.

TimeO(log n) amortized per event

SpaceO(events within lateness window)

tests.py · 4 passing

tests.pypython9 lines

d = StreamDeduplicator(allowed_lateness=10)
stream = [("a", 1), ("b", 2), ("a", 1), ("c", 5), ("b", 2)]
assert [k for k, t in stream if d.process(k, t)] == ["a", "b", "c"]
assert d.duplicates == 2
for t in range(100, 1100):
    d.process(f"k{t}", t)
    d.process(f"k{t}", t)  # retry delivered twice
assert d.state_size() <= 12 and d.duplicates == 1002
assert d.process("a", 1) is False and d.late == 1  # too late, not re-emitted

04Staff-level follow-up

How do you get exactly-once end to end across a crash? Checkpoint the dedupe state and source offsets atomically (Flink/Spark checkpoint), and make the sink idempotent or transactional (upsert by key, or two-phase commit tied to the checkpoint).