Data Reliability and Fault Tolerance Questions

Direct answer

An operation is idempotent when applying it once, or applying it N times with the same input, leaves the system in the same state as applying it exactly once. This matters for pipelines because retries and event replays are inevitable (network timeouts, consumer restarts, at-least-once delivery), and every one of those retries resends work whose first attempt may already have succeeded silently. Without idempotency the pipeline cannot tell "this is new work" from "this is a duplicate of work already done," so retries silently corrupt state: duplicate rows, double-counted aggregates, or a payment charged twice. Three concrete sink-level patterns make writes idempotent: keyed upsert/merge with a version or timestamp, transactional writes with atomic commit, and content-addressable or object-versioned writes.

Structured elaboration

Why "retry-safe" is not automatically "idempotent." A naive retry (call the same INSERT again) is retry-safe in the sense that it does not crash, but it is not idempotent: it produces a second row. Idempotency requires the sink to recognize a repeated logical operation and either no-op it or converge to the same result. That recognition needs a stable key that identifies the logical operation (a business key, an event ID, or a derived content hash), not just the physical retry.

Pattern 1: Upsert or merge by primary key with a version or timestamp.

• When it applies: row-oriented sinks with a natural primary key and a native conditional-write or MERGE operation: relational databases, Cassandra, DynamoDB, or table formats like Delta Lake / Apache Iceberg that support MERGE INTO.
• Mechanism: write INSERT ... ON CONFLICT (pk) DO UPDATE WHERE incoming.version > existing.version (or timestamp-based equivalent). A retry with the same key and same or older version either no-ops or overwrites with an identical value, so the end state is unchanged.
• Trade-offs: cost is low, this is standard database machinery. Latency is low for single-row writes, but rises under batch upserts because concurrent writers to the same key contend on row locks. Complexity is medium: you need a monotonic version or timestamp source (a wall clock alone is unsafe under clock skew; a source-system sequence number or Lamport-style counter is safer), and you must define tie-breaking for equal versions.

Pattern 2: Transactional writes with atomic commit.

• When it applies: when a single logical write must land atomically across multiple partitions, tables, or heterogeneous objects, so a partial write is unacceptable. Examples: multi-row OLTP transactions, or a data-lake write that must atomically add multiple Parquet files plus a manifest update (Delta/Iceberg's commit protocol).
• Mechanism: wrap the write in a transaction (or a two-phase commit / atomic manifest swap) keyed by an idempotency token, so a retried transaction with the same token is rejected or recognized as already-applied by the coordinator before any partial state is visible to readers.
• Trade-offs: cost is higher (a transaction coordinator, locks, or an atomic-rename/manifest layer). Latency is higher because the commit protocol adds round trips and the writer must wait for confirmation before releasing correctness guarantees. Complexity is high: the coordinator itself needs to be crash-recoverable, and you must handle the case where the commit succeeded but the acknowledgment was lost (which is exactly what makes retries dangerous in the first place).

Pattern 3: Content-addressable or object-versioned writes.

• When it applies: immutable, append-only sinks such as object storage (S3, GCS) holding blobs or files, where rewriting in place is undesirable or impossible.
• Mechanism: derive the object's key (or a companion manifest entry) from a hash of its content, or from a caller-supplied idempotency token used as the object version. Writing the same content twice produces the same key, so a retry is a no-op PUT to an already-existing key rather than a second distinct object.
• Trade-offs: cost includes extra storage for a hash index or version metadata and eventual garbage collection of superseded versions. Latency for the write itself is low (a PUT is a PUT), but a pre-write existence check or post-write reconciliation adds overhead. Complexity is medium: hashing must be deterministic (careful with non-deterministic serialization, e.g. dict key ordering or float formatting), and this pattern only naturally handles create/append; in-place updates need a separate mechanism.

Worked example

Take a payment-charge API called from a pipeline step, with no idempotency mechanism. Assume the server successfully commits the charge but the acknowledgment is lost on the wire 2% of the time (a realistic order of magnitude for transient network faults), and the client's retry policy blindly resends on any timeout. Over 1,000,000 transactions:

At an average charge of $50, that is:

Adding an idempotency key (pattern 1 or 2 above: the client generates a UUID per logical charge attempt, the server upserts on that key) converts every one of those 20,000 retries into a no-op that returns the original charge's result. The duplicate-charge count drops to (in principle) zero, bounded only by the idempotency key's own storage durability, not by network reliability. The same arithmetic applies directly to duplicate charges and double-counted metrics as the concrete production symptom of a missing idempotency key: a metrics pipeline with the same 2% duplicate-delivery rate and no dedup key would silently inflate every downstream count (revenue, event volume) by roughly 2%, which is large enough to distort a dashboard but subtle enough that nobody notices until finance or product asks why the numbers do not reconcile. The same failure shows up in ETL form too: an ingestion job that retries a failed micro-batch without an idempotency key re-appends the successfully-written rows from the failed batch's completed prefix, corrupting downstream aggregates the same way.

Trade-offs and pitfalls

• Combining patterns is normal, not a code smell. A common production shape is pattern 2 (transactional commit) for the manifest plus pattern 3 (content-addressed files) for the underlying data: the files are naturally idempotent to re-upload, and the transactional manifest swap makes their visibility atomic.
• Common mistake: assuming idempotency is free once a key exists. The idempotency key itself must be generated deterministically from the logical operation (not from uuid4() at retry time, which produces a new key on every retry and defeats the whole mechanism) and the key-lookup store must itself survive a crash, or the pipeline just moved the duplication problem one layer down.
• Common mistake: unbounded idempotency-key retention. Keeping every key forever grows storage without bound; keeping it too briefly (a short TTL) reopens the duplicate window for retries that arrive late (e.g. after a long consumer pause). The retention window must be sized to the pipeline's actual maximum retry delay, not a default.
• Cost/latency/complexity is a real spectrum, not a checkbox. Pattern 1 is usually the cheapest and should be the default; reach for pattern 2 only when atomicity spans multiple objects, and pattern 3 only for genuinely immutable/append-only data, since forcing an update-heavy workload through content-addressing means constantly writing new versions and garbage-collecting old ones.

Direct answer

An idempotent POST handler needs three things: a client-supplied request_id that uniquely names the logical operation, a store that maps that ID to the response it already produced, and an atomic check-then-set so two retries racing in at once cannot both execute the side effect. Below is a plain WSGI handler (no framework dependency) backed by a lock-protected dictionary with per-entry TTL. Dict lookups are O(1) average case, and the lock is held across the whole check-compute-set sequence, which is the detail that actually makes it concurrency-safe rather than merely "looks idempotent in the happy path."

Structured elaboration

Why check-then-set must be one atomic step, not two. The tempting bug is: check the dict for request_id under the lock, release the lock, then compute and store the result. That reopens exactly the race the handler exists to close: two threads can both see "not present" before either has written its result, and both run the side effect. The lock in the implementation below stays held across compute_fn(), so a second thread arriving mid-computation blocks until the first thread's result is visible, then reads that cached result instead of recomputing.

TTL is a memory-bound, not a correctness guarantee. The store cannot grow forever, so entries expire after ttl_seconds. This means the handler is idempotent only within the TTL window: a retry that arrives after the entry expired is treated as a new operation and reruns the side effect. The TTL must therefore be set to comfortably exceed the caller's maximum retry backoff, not to an arbitrary "cache expiry" value borrowed from an unrelated system.

O(1) lookup, not O(1) memory. A plain Python dict gives average O(1) get/set, but the size of the dict grows with the number of distinct in-flight request IDs. Production versions add either a bounded LRU eviction on top of the TTL, or move the store to something with native TTL and eviction (Redis SETEX, DynamoDB with a TTL attribute) so the process does not accumulate unbounded memory between GC/purge cycles.

Worked example

"""
Idempotent HTTP POST handler using plain WSGI (no Flask dependency needed).
Demonstrates: in-memory store with TTL, thread-safe (lock-protected) O(1) lookups,
returns the cached response for a repeated request_id.
"""
import json
import time
import threading


class IdempotencyStore:
    """Thread-safe in-memory cache: request_id -> (response_body, expires_at)."""

    def __init__(self, ttl_seconds=300):
        self._ttl = ttl_seconds
        self._lock = threading.Lock()
        self._store = {}  # dict lookups are O(1) average case

    def get_or_set(self, request_id, compute_fn):
        """
        Atomic check-then-set under ONE lock so two concurrent requests with the
        same request_id cannot both "win" and run compute_fn twice.
        """
        now = time.time()
        with self._lock:
            entry = self._store.get(request_id)
            if entry is not None:
                response, expires_at = entry
                if expires_at > now:
                    return response, True
                # expired: fall through and recompute
            # Compute WHILE STILL HOLDING THE LOCK, so a racing thread blocks
            # here instead of also seeing "not cached".
            response = compute_fn()
            self._store[request_id] = (response, now + self._ttl)
            return response, False

    def purge_expired(self):
        now = time.time()
        with self._lock:
            expired = [k for k, (_, exp) in self._store.items() if exp <= now]
            for k in expired:
                del self._store[k]
            return len(expired)

    def size(self):
        with self._lock:
            return len(self._store)


STORE = IdempotencyStore(ttl_seconds=300)
CHARGE_COUNTER = {"n": 0}
COUNTER_LOCK = threading.Lock()


def process_payload(payload):
    """Simulates the actual side-effecting work (e.g., charging a card)."""
    with COUNTER_LOCK:
        CHARGE_COUNTER["n"] += 1
        return {"status": "processed", "charge_seq": CHARGE_COUNTER["n"], "echo": payload}


def application(environ, start_response):
    if environ.get("REQUEST_METHOD") != "POST":
        start_response("405 Method Not Allowed", [("Content-Type", "application/json")])
        return [json.dumps({"error": "POST only"}).encode()]
    try:
        length = int(environ.get("CONTENT_LENGTH", 0) or 0)
        raw = environ["wsgi.input"].read(length)
        body = json.loads(raw)
        request_id = body["request_id"]
        payload = body["payload"]
    except (KeyError, ValueError, TypeError):
        start_response("400 Bad Request", [("Content-Type", "application/json")])
        return [json.dumps({"error": "request_id and payload required"}).encode()]

    response, was_cached = STORE.get_or_set(request_id, lambda: process_payload(payload))
    headers = [("Content-Type", "application/json"),
               ("X-Idempotent-Replayed", "true" if was_cached else "false")]
    start_response("200 OK", headers)
    return [json.dumps(response).encode()]


def call_app(request_id, payload):
    import io
    body = json.dumps({"request_id": request_id, "payload": payload}).encode()
    environ = {"REQUEST_METHOD": "POST", "CONTENT_LENGTH": str(len(body)), "wsgi.input": io.BytesIO(body)}
    captured = {}
    def start_response(status, headers):
        captured["status"] = status
        captured["headers"] = dict(headers)
    result = application(environ, start_response)
    return captured["status"], captured["headers"], json.loads(b"".join(result))


def main():
    s1, h1, b1 = call_app("abc-1", {"amount": 42})
    print("Call 1:", s1, h1["X-Idempotent-Replayed"], b1)

    s2, h2, b2 = call_app("abc-1", {"amount": 42})  # retry, same request_id
    print("Call 2 (retry):", s2, h2["X-Idempotent-Replayed"], b2)

    s3, h3, b3 = call_app("abc-2", {"amount": 99})  # different request_id
    print("Call 3 (new id):", s3, h3["X-Idempotent-Replayed"], b3)

    # Concurrency proof: 20 threads hit the SAME new request_id at once.
    results, results_lock = [], threading.Lock()
    def worker():
        _, _, b = call_app("concurrent-1", {"amount": 7})
        with results_lock:
            results.append(b["charge_seq"])
    threads = [threading.Thread(target=worker) for _ in range(20)]
    for t in threads: t.start()
    for t in threads: t.join()
    print("Concurrency test: 20 threads, same request_id, distinct charge_seq values:", sorted(set(results)))
    print("Total side-effect executions across the run:", CHARGE_COUNTER["n"])


if __name__ == "__main__":
    main()

Output (actually executed with python3):

Call 1: 200 OK false {'status': 'processed', 'charge_seq': 1, 'echo': {'amount': 42}}
Call 2 (retry): 200 OK true {'status': 'processed', 'charge_seq': 1, 'echo': {'amount': 42}}
Call 3 (new id): 200 OK false {'status': 'processed', 'charge_seq': 2, 'echo': {'amount': 99}}
Concurrency test: 20 threads, same request_id, distinct charge_seq values: [3]
Total side-effect executions across the run: 3

Call 2 replays call 1's exact result (X-Idempotent-Replayed: true, same charge_seq: 1, no new charge). The concurrency test is the load-bearing proof: 20 threads racing on the identical request_id produce exactly one distinct charge_seq and the global counter shows exactly 3 total executions across the whole run (call 1, call 3, and the single winner of the 20-way race), not 22. That is what "concurrency-safe" concretely means here: the lock, not luck, decided the outcome.

Trade-offs and pitfalls

• This is not a hypothetical bug. Dropping the lock between check and set breaks the guarantee, and it is worth seeing fail, not just asserting:

import threading, time

class BrokenStore:
    def __init__(self):
        self._lock = threading.Lock()
        self._store = {}
    def get_or_set(self, request_id, compute_fn):
        with self._lock:
            entry = self._store.get(request_id)
            if entry is not None:
                return entry, True
        time.sleep(0.001)  # widen the check-to-set race window
        response = compute_fn()
        with self._lock:
            self._store[request_id] = response
        return response, False

COUNTER = {"n": 0}
lock = threading.Lock()
def process():
    with lock:
        COUNTER["n"] += 1
        return COUNTER["n"]

store = BrokenStore()
results, rl = [], threading.Lock()
def worker():
    r, _ = store.get_or_set("same-id", process)
    with rl:
        results.append(r)
threads = [threading.Thread(target=worker) for _ in range(20)]
for t in threads: t.start()
for t in threads: t.join()
print("Broken store, 20 threads, same request_id, distinct results:", sorted(set(results)))
print("Total side-effect executions:", COUNTER["n"])

Output (actually executed with python3):

Broken store, 20 threads, same request_id, distinct results: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20]
Total side-effect executions: 20

All 20 threads ran the side effect independently, exactly the duplicate-charge bug idempotency exists to prevent. The only structural difference from the working version is where the lock is released.

• Common mistake: client mints a new request_id on every retry (e.g. uuid4() generated fresh in the retry loop instead of reused from the original attempt). The handler is correct but useless if the caller does not reuse the ID.
• Common mistake: no TTL at all. An unbounded dict is a slow memory leak in a long-running process; TTL plus periodic purge_expired() (or moving to a store with native TTL like Redis or DynamoDB) bounds it.
• This in-memory store does not survive a process restart or scale past one process. In production, the store itself needs to be externalized (Redis, DynamoDB) so idempotency holds across restarts and horizontally scaled instances; the in-memory version here is correct for a single process and is the right shape to translate directly onto an external store.

Direct answer

The right choice is workload-driven, not universally "better": an analytics pipeline should default to at-least-once processing plus deduplication, because analytics tolerates small windows of eventual correctness in exchange for much simpler operations and higher throughput, while a payments pipeline should pay for low-latency exactly-once (Kafka transactions plus Flink's two-phase-commit sink, or an equivalent) because a financial system cannot tolerate even a transient double-charge, and its lower relative throughput requirement makes the extra operational cost affordable.

Structured elaboration

Throughput. Kafka/Flink transactional exactly-once adds coordination overhead: a transaction commit protocol across producer, broker, and consumer offsets, plus (for Flink) checkpoint-aligned barriers gating when transactions can commit. This measurably reduces maximum sustainable throughput compared to plain at-least-once (no coordination, just produce and consume as fast as the hardware allows). At-least-once plus downstream dedup keeps the hot path simple and fast, paying the correctness cost later, at the dedup step, which can often be batched or made cheaper than transactional coordination on every single record.

Complexity. Exactly-once via transactions requires every component in the chain to participate correctly: idempotent producers, transactional consumers reading only committed offsets (read_committed isolation level), and checkpoint-aligned sink commits. Misconfiguring any one piece (an accidental read_uncommitted consumer, a non-transactional sink in the chain) silently breaks the guarantee without any error, since the pipeline keeps running, just with a weaker guarantee than intended. At-least-once-plus-dedup is conceptually simpler: get everything delivered, then deduplicate by key, a technique any single team can implement and reason about locally.

Operational burden. Exactly-once transactional pipelines need careful monitoring of transaction timeouts (a stalled transaction blocks downstream consumers reading read_committed, since they cannot see past an open transaction) and of checkpoint health (a stalled checkpoint stalls transaction commits). At-least-once-plus-dedup's operational surface is simpler: monitor consumer lag and dedup-store health, both well-understood, widely-tooled concerns.

Failure scenarios. Under exactly-once transactions, a coordinator or broker failure mid-transaction can leave a transaction open longer than its timeout, causing downstream consumers to stall waiting for it to resolve, a availability-for-consistency trade that is exactly the E in CAP terms being spent. Under at-least-once-plus-dedup, the same failure just causes redelivery, handled by the dedup layer with no consumer-side stall, at the cost of a brief window where duplicates could theoretically reach a consumer that skips dedup (a real risk only if the dedup step itself is bypassed or misconfigured).

Worked example

Analytics pipeline: 500,000 events/sec of clickstream data, at-least-once ingestion via Kafka, deduplicated downstream using a 24-hour sliding window on event_id. A brief network blip causing 0.1% redelivery:

No consumer stall, no transaction timeout risk; the analytics aggregate is correct within the normal dedup-window latency, and 500,000 events/sec throughput is realistically sustainable with plain at-least-once delivery.

Payments pipeline: 2,000 transactions/sec (a full order of magnitude lower than the analytics case, typical of a payments-critical path versus a clickstream), using Kafka transactions end to end with a transactional, idempotent sink (a Flink job with exactly-once checkpointing and a two-phase-commit database sink). At 2,000 TPS, the added coordination overhead of transactional commits is comfortably absorbable, whereas at 500,000 events/sec the same coordination overhead would likely become the pipeline's throughput ceiling. This is the concrete shape of "throughput budget determines whether the exactly-once tax is affordable": payments pays it because it can and must; analytics does not pay it because the value of doing so is lower and the cost of paying it (throughput ceiling) is higher relative to its actual load.

Trade-offs and pitfalls

• Common mistake: assuming exactly-once via transactions eliminates the need for any dedup logic at all. Transactional exactly-once guarantees correctness WITHIN the transactional chain (producer to this specific consumer's committed offset), not necessarily end-to-end if any hop in the chain is not itself transactional (a REST call to a legacy system with no transaction support); a payments pipeline calling such a system still needs idempotent-write discipline at that hop.
• Common mistake: choosing at-least-once-plus-dedup for payments "because it's simpler." Simplicity is a real virtue, but for payments the downside of a dedup-layer bug (a transient double-charge visible to the customer before correction) is categorically worse than the operational cost of transactions; the choice should be driven by the cost of a guarantee failure, not by engineering convenience alone.
• Transaction timeouts are a real, underestimated operational risk. A payments team that adopts exactly-once transactions without also monitoring and alerting on transaction duration can be surprised by a slow downstream dependency silently stalling every consumer reading read_committed, a failure mode that simply does not exist under at-least-once-plus-dedup.
• The choice is not binary across an entire pipeline. A hybrid is common and often correct: exactly-once for the specific payments-critical hop, at-least-once-plus-dedup for a downstream analytics fan-out reading the same event stream for reporting, since the reporting consumer's correctness needs do not justify inheriting the payments hop's operational cost.

Direct answer

Combining streaming CDC with periodic batch replays for exactly-once semantics requires one authoritative ordering key (the source's LSN or transaction ID) that BOTH paths use identically, and a target write pattern (an idempotent, versioned MERGE) that makes it irrelevant whether a given change arrived via the streaming path or a batch replay, since both converge to the same guarded upsert. The critical design point is defining, precisely, which LSN range each batch replay covers and recording that range durably, so streaming and batch never silently reprocess (or silently skip) the same range, and the reconciliation between them is a bounded, explicit check, not an assumption.

Structured elaboration

Why batch replays exist alongside streaming CDC at all. Common real reasons: a historical backfill when the pipeline is first stood up (before streaming CDC existed, the OLTP system's full history needs one batch load), periodic full-table reconciliation to catch anything streaming CDC might have missed (a connector bug, a period the connector was down longer than its retention window), or a deliberate large-scale reprocessing after a downstream bug fix. In every case, the SAME target table now receives writes from two different pipelines that must agree on ordering and idempotency.

Ordering. Every batch-replayed row must carry (or be joined against) the same source LSN/transaction-ID metadata the streaming path uses, typically by reading it directly from the source (a snapshot query that includes the row's last-modified transaction ID) or, if the source does not expose this per-row, by treating the entire batch as having a single, coarse "as-of" LSN corresponding to when the batch snapshot was taken. The finer-grained per-row LSN is strongly preferable when available, since it lets the batch path participate in the exact same fine-grained ordering guard as streaming, rather than a coarse all-or-nothing batch-level ordering.

Idempotency. The target's MERGE guard (apply only if incoming.lsn > target.lsn) needs no modification to also handle batch replay input, as long as batch rows carry a real LSN. This is the core design insight: idempotency and ordering are properties of the TARGET's apply logic, not of which upstream path produced the input, so streaming and batch can share the identical apply step rather than needing separate reconciliation code paths.

Transactional boundaries. A source-side multi-row transaction (e.g. a transfer that debits one account and credits another) must be reflected atomically at the target, both changes visible together or neither. Streaming CDC connectors typically preserve transaction boundaries as metadata (Debezium emits a transaction ID per event and, optionally, transaction begin/end markers); a batch replay reading a point-in-time snapshot naturally sees only fully-committed transactions as of the snapshot instant (assuming a consistent, transactionally-isolated snapshot read, e.g. Postgres's repeatable-read or a logical replication slot's consistent snapshot), so both paths preserve atomicity, streaming via explicit transaction metadata, batch via snapshot isolation, different mechanisms achieving the same property.

Reconciling streaming events with batch backfills without double-counting. Because the target apply logic is idempotent and LSN-guarded, "double-counting" in the sense of applying the same LSN twice is already prevented. The remaining risk is a COVERAGE gap or overlap: if the batch replay's snapshot boundary and the streaming path's LSN range are not explicitly recorded and compared, you cannot verify whether a specific LSN range was covered by exactly one of the two paths (correct), by both (harmless, since idempotent, but wasteful), or by NEITHER (a real, silent gap). Recording, per batch run, the exact LSN range it covers (its start and end position in the source log) alongside the streaming connector's own continuously-advancing LSN watermark, and periodically diffing the two, turns "did we lose data in the seam" from an assumption into a checkable, alertable fact.

Worked example

A Postgres OLTP source streams CDC continuously via Debezium, currently at LSN watermark 5,000,000. A batch reconciliation job runs weekly, taking a consistent snapshot as of a specific point, recorded as "snapshot covers all committed transactions up to LSN 4,995,000" (a few thousand LSNs behind the live streaming watermark, since the snapshot was initiated slightly earlier and took time to complete). The batch job applies its rows through the identical MERGE-guard logic used by streaming.

For any given row, three cases are possible, and the guard resolves all three correctly:

1. Streaming already applied the row at LSN 4,994,500 (before the snapshot boundary), and the batch replay also includes it at the same LSN 4,994,500: 4{,}994{,}500 > 4{,}994{,}500 is false, batch's application is a no-op, target unaffected. Correctly handles overlap without double-counting.
2. Streaming has NOT yet applied a row that changed at LSN 4,994,800 (still in its consumer lag, hypothetically behind), and the batch replay includes it (since 4,994,800 < 4,995,000, the snapshot boundary): batch applies it first, target now at lsn=4,994,800; when streaming later catches up and attempts to apply that same LSN, the guard correctly no-ops it. Correctly handles either path "winning the race" to apply a given change first.
3. A row changed at LSN 4,995,500 (after the snapshot boundary): the batch snapshot does not include it at all (it queried state as of an earlier point), so only streaming applies it, whenever streaming's consumer catches up to that LSN. No double-counting risk since only one path ever sees it.

The recorded batch coverage boundary (4,995,000) versus the streaming watermark (5,000,000, and continuously advancing) is exactly the pair of numbers a reconciliation check compares: as long as streaming's watermark is confirmed to have passed 4,995,000 with no detected connector gap in between, the two paths are known to jointly cover the full LSN range with no silent hole, which is the concrete, checkable version of "reconcile without double-counting" the question asks for.

Trade-offs and pitfalls

• Common mistake: giving the batch path a SEPARATE apply/merge implementation from streaming, reasoning "batch is simpler, it doesn't need the guard, we control when it runs." This is exactly what reopens double-counting or silent overlap risk; the whole design's strength comes from both paths sharing one idempotent, LSN-guarded apply step.
• Common mistake: recording the batch snapshot's boundary imprecisely (e.g. "ran around 2am") instead of the actual source LSN as of the transactionally-consistent snapshot instant. A wall-clock approximation cannot be compared against the streaming watermark's own LSN units, making the coverage reconciliation described above impossible to actually perform.
• Transaction-boundary preservation is easy to lose silently. If the batch snapshot's isolation level is weaker than repeatable-read (or the source does not support a truly consistent point-in-time read), the batch may capture one side of a multi-row transaction but not the other, breaking atomicity in a way the LSN guard alone does not detect, since each row's LSN guard passes independently, unaware that they were meant to arrive together.
• This design assumes the source retains enough transaction-log history to bridge any gap between "streaming's current watermark" and "the batch snapshot's boundary." If the source's log retention is shorter than the realistic time between a connector outage and its detection, that portion of history is permanently unrecoverable by either path, which argues for the reconciliation check running frequently enough (or the log retention being generous enough) that no such gap can silently exceed the retention window.

Direct answer

When the third-party API itself cannot be made idempotent (no idempotency-key parameter, no way to query "did this already happen"), the fix has to move up a layer: your own system becomes the source of idempotency, and the API call becomes just one, possibly-retried, step inside a larger idempotent workflow. Three system-level patterns do this: the outbox pattern (durably record intent to call before calling, so a crash mid-call is always recoverable), a dedicated idempotency proxy (a thin service in front of the third-party API that adds the idempotency the API itself lacks), and compensated eventual-consistency workflows (accept that a duplicate or failed call can happen, and design an explicit compensating step to detect and correct it after the fact).

Structured elaboration

Why the API being at-most-once (not the pipeline's choice) changes the problem. With an idempotent API, retries are free: send the same request with the same key, get the same result. With a strictly at-most-once, non-idempotent API, every retry is a genuine risk of a second real effect (a second charge, a second email sent), and every NON-retry after an ambiguous failure (timeout with unknown server-side outcome) is a genuine risk of a missed effect. You cannot fix this by changing your own retry logic alone; you need a mechanism to know, independently of the API's own response, whether the call already happened.

Pattern 1: The outbox pattern.

• Mechanism: within the same local transaction as the business logic that decides "call the API," write a row to an outbox table recording the intended call and a unique operation ID. A separate, durable process reads unconfirmed outbox rows and calls the API, marking the row confirmed only after getting a definitive success response. On crash before confirmation, the outbox row is still there on restart, so the (still-pending) call is retried; a crash AFTER a successful call but before marking it confirmed risks exactly one duplicate call, which the outbox pattern alone does not prevent, only bounds to "at most once more."
• Trade-offs: cheap and simple to build on top of any transactional local database; does not by itself solve the ambiguous-timeout case (the actual API call still might have succeeded silently), only guarantees the ATTEMPT is never silently lost.

Pattern 2: A dedicated idempotency proxy.

• Mechanism: introduce a small internal service that sits between your pipeline and the third-party API. Your pipeline calls the proxy with a caller-generated idempotency key; the proxy checks its own durable store for that key. If seen before, it returns the previously recorded outcome without calling the API again. If not seen, it calls the API, durably records the outcome (success, failure, or genuinely unknown, e.g. after its own timeout waiting on the API), and returns that outcome, all before considering the operation complete, so the proxy's own record becomes the authoritative "did this happen" answer that the third-party API cannot provide.
• Trade-offs: this actually solves the ambiguous-timeout case (the proxy's durable record, not the API's response, is the source of truth), at the cost of building and operating an extra stateful service, and the proxy's own store needs the same crash-recoverability discipline (a WAL, a transactional database) it is providing to everything downstream of it.

Pattern 3: Compensated eventual-consistency workflows.

• Mechanism: accept the API may be called more than once (or, in some rare failure interleavings, produce an effect you cannot detect), and instead of preventing the duplicate, detect it after the fact (reconciling against the API's own downstream records, if it exposes any, such as a transaction history endpoint) and issue a compensating action (a cancellation, a refund, a corrective follow-up call) when a duplicate is found.
• Trade-offs: this is the fallback when the API supports neither a native idempotency mechanism NOR any reasonable proxy-level prevention (e.g., a one-way SMS API with no way to check delivery history at all); it accepts a real, sometimes user-visible window of duplication in exchange for being implementable against literally any API, however primitive.

Worked example

A pipeline calls a legacy shipping-label API (no idempotency key support) to generate a shipping label and charge a small label fee, and observes a 1.5% rate of ambiguous timeouts (request sent, no response received, unknown whether the label was actually generated) across 200,000 calls/month:

Naive retry-on-timeout: all 3,000 are blindly retried, and if roughly half of those ambiguous timeouts were ACTUALLY successful server-side (a stated assumption; the true fraction is unknown by definition, which is exactly the problem), that is:

With an idempotency proxy in front of the API: the proxy durably records "call in flight" before calling, and on the ambiguous timeout, the retry goes to the PROXY (which already has a durable record of an in-flight, outcome-unknown call), not directly to the API. The proxy's own policy (poll the API's separate order-status endpoint if one exists, or apply a conservative "treat unknown as done, require manual reconciliation" rule if it does not) replaces "duplicate call happens 1,500 times a month" with "1,500 calls per month go into an explicit reconciliation queue," a bounded, observable, and fixable number instead of a silent, ongoing revenue and customer-experience leak.

Trade-offs and pitfalls

• Common mistake: treating a timeout as either 'definitely failed, retry' or 'definitely succeeded, don't retry.' Neither is safe for a non-idempotent API; the honest state is "unknown," and the system-level pattern exists precisely to make that unknown state explicit and durably tracked rather than guessed at.
• The outbox pattern alone does not fully close the gap. It guarantees the ATTEMPT is never lost, but a crash between "API call succeeded" and "outbox row marked confirmed" still risks exactly one duplicate retry on restart; an idempotency proxy (pattern 2) or a reconciliation step (pattern 3) is needed to close that remaining, narrower window.
• Building a proxy is a real engineering investment, not a quick fix. It is worth it when call volume and the cost of duplication (a real financial charge, a real physical shipment) justify a dedicated stateful service; for a low-volume, low-stakes integration, accepting the compensating-transaction fallback may be the more proportionate choice.
• Compensating actions need their own idempotency discipline. A compensation issued twice for the same detected duplicate (a double refund) is the exact same class of bug this whole answer exists to prevent, just one layer further out; the reconciliation/compensation step must track which duplicates it has already compensated.