How to Prove a Distributed Pipeline Is Retry-Safe

In F04.2, we learned that a task can successfully write results to an external database and still be retried. A lost completion acknowledgement may cause the coordinator to execute the same publish task again. We also learned that idempotent writes can prevent these retries from creating duplicate ledger records. But there’s another problem we haven’t considered.

What if the input data itself contains repeated deliveries of the same event?

Even if every processing task runs exactly once and every database write is idempotent, the calculation can still be incorrect. Let’s understand why.

The Order Feed Has Changed

Our nightly ledger job processes events from an order management system. Previously, the source delivered each event once.

For example:

Each event appeared once, and our revenue aggregation simply added the order amounts for each customer. Now the source team introduces a new delivery mechanism. The updated system uses at-least-once delivery. This means it retries delivery when it cannot confirm that an event was successfully received.

For example:

1. 01The source sends an order event to the receiving system.
2. 02The receiving system accepts the event.
3. 03The acknowledgment is lost before reaching the source.
4. 04The source assumes delivery may have failed.
5. 05It sends the same event again.

The receiving system now sees two deliveries of the same logical event. This should sound familiar. It’s the same uncertainty we studied in F04.2, but in the opposite direction. In F04.2, the coordinator repeated a database write because a completion message was lost. Here, the source repeats an event delivery because its acknowledgment was lost.

An Event Is Not the Same as a Delivery

This distinction is central to the lesson. An event represents something that happened in the business. A delivery represents one attempt to communicate that event to another system. The same event may be delivered more than once.

For example:

Notice the first and third rows. They have different delivery_id values but the same event_id. This means:

• del-001 is the first delivery of event evt-101.
• del-003 is another delivery of the same event.

There are three deliveries but only two distinct events. If we calculate revenue by adding the amounts from every delivery, evt-101 contributes 2,000 cents instead of 1,000 cents. The aggregation becomes incorrect even though no new business event occurred.

What Does Our New Dataset Contain?

Let’s examine the actual fixture used in this lesson.

The difference between total deliveries and distinct events is: 22,363 − 21,198 = 1,165 additional deliveries. These represent repeated arrivals beyond the first delivery of each event. Notice that 1,058 events were delivered at least twice, but there are 1,165 additional deliveries. Why are those numbers different? Because some events arrived more than twice.

For example, if one event is delivered four times, it contributes:

• One distinct event.
• Four total deliveries.
• Three additional deliveries.

This explains why the number of additional deliveries can exceed the number of distinct events that were redelivered.

The Source Also Contains Corrections

There’s another complication. The new feed contains 44 correction events. A correction is different from a redelivery. A redelivery represents another arrival of the same event. A correction is a new event that changes or replaces an earlier event. For example, suppose an order was initially recorded with an amount of 1,000 cents. Later, the source discovers that the correct amount should be 1,200 cents. Rather than modifying the original event, it produces a new correction event. For our lesson, the business contract says that the correction replaces the original event.

We can represent this as:

Here, evt-202 is a new event, but it replaces evt-201. The correct contribution to revenue is 1,200 cents, not 2,200 cents. If our aggregation simply adds every event, it will count both amounts.

Three Fields That Define the Correct Behaviour

Our source records include three important identifiers.

These fields serve different purposes. event_id helps identify repeated deliveries of the same business event. delivery_id helps track individual arrivals. supersedes_event_id helps determine which earlier event should no longer contribute to the final result. Confusing these identities can produce incorrect aggregations.

Why Didn’t We See This Problem Before?

Our previous fixture contained:

• 21,154 deliveries
• 21,154 distinct events
• No corrections
• 797 customers

Every event was delivered exactly once. As a result, adding the amounts from every delivery produced the correct revenue total. But that correctness depended on an assumption: one delivery always represented one event. The new source no longer guarantees that assumption. The aggregation code hasn’t changed, but the input behaviour has.

This is a common source of production failures. A pipeline can produce correct results for months because the input happens to satisfy assumptions that were never explicitly enforced. When the source’s behaviour changes, those hidden assumptions become bugs.

What About Late-Arriving Events?

Our fixture also contains 217 deliveries that arrived after the reporting window closed. The nightly ledger uses a fixed reporting cutoff. Only deliveries admitted by that cutoff are considered for the current run. Late deliveries are handled according to a subsequent processing window or correction policy rather than silently changing an already completed result.

This follows the deterministic-processing principle we studied in F04.1.

Now let’s run our existing revenue aggregation against the new input feed. In F03, we learned how to calculate customer revenue using a distributed sum. Each map task calculates a partial revenue total for its customers. The reduce stage combines those partial totals. The underlying aggregation is straightforward:
Customer revenue = Sum of the order amounts assigned to that customer

This works when every qualifying order contributes exactly once. But our new feed can deliver the same order event multiple times and contains corrections that replace earlier events. What happens if we keep summing every delivery?

Understanding the Existing Aggregation

Our current processing state stores one running total per customer. Conceptually, the operation looks like this: customer_revenue[customer_id] += amount_cents

Every qualifying delivery contributes its amount to the customer’s total. Consider this simplified example.

Suppose evt-102 is a correction that replaces evt-101. The correct revenue contribution is: 1,200 cents

However, the current aggregation adds all three deliveries: 1,000 + 1,000 + 1,200 = 3,200 cents

There are two separate errors:

1. 01The redelivery of evt-101 adds another 1,000 cents.
2. 02The correction evt-102 is added without removing the event it replaces.

A sum doesn’t know whether a record is a new event, a repeated delivery, or a replacement. It only knows how to add values.

Our new feed contains:

• 22,363 deliveries
• 21,198 distinct events
• Repeated deliveries of some order events.
• Correction events that replace earlier events.

The correct revenue total, based on the business contract, is: 74,558,550 cents. However, the existing job simply sums the amounts from every qualifying delivery.

Why Is the Total Incorrect?

The problem comes from two kinds of input records.

1. Redelivered order events
The new feed contains 219 order events that were delivered more than once. The sum processes each delivery independently. Therefore, a redelivered order may contribute its amount multiple times.

2. Correction events
The feed also contains 44 correction events. Under our contract, each correction replaces an earlier event. However, the sum doesn’t remove the amount belonging to the superseded event. It simply adds the correction’s amount. Both behaviours increase the calculated revenue relative to the contract truth in this fixture.

Let’s Examine One Customer

Consider cust-0001. The correct revenue for this customer is: 5,441,415 cents
But the current aggregation produces: 5,962,004 cents
The difference is: 5,962,004 − 5,441,415 = 520,589 cents

This customer’s revenue is overstated by 520,589 cents because repeated order deliveries contribute amounts more than once. The same issue affects other customers, although the size of the error varies according to their redeliveries and corrections. Across the complete fixture, 164 customer totals are incorrect.

Why Did the Same Aggregation Work Last Month?

Let’s compare the new feed with the previous one.

The previous fixture contained no repeated deliveries or corrections. Therefore, summing every delivery happened to be equivalent to summing the correct logical events. The new feed breaks that equivalence. The same code now produces incorrect results because the source’s delivery semantics have changed.

Can We Fix This by Adding More Workers?

No. This isn’t a performance problem. Adding workers might reduce execution time, but it won’t change the meaning of the aggregation. The job would still process the same duplicate deliveries and correction events. Similarly, changing the number of partitions or reduce tasks would not resolve the error. The issue is with the processing state and the business rules it represents.

We need an aggregation that understands event identity and correction relationships. Before redesigning the entire pipeline, however, the team tries a simpler solution: deduplication.

The team investigates why the revenue increased after the source-system upgrade. The source documentation explains: The feed uses at-least-once delivery. If an event isn’t acknowledged, it may be delivered again. Every delivery attempt receives a new delivery_id.

The engineers decide to add a deduplication step. Their reasoning is straightforward:
If the problem is duplicate records, remove duplicates before calculating revenue. They implement a filter that keeps the first record for each delivery_id and discards any subsequent records with the same ID. Will this solve the problem?

Understanding What the Patch Actually Does

Consider the following source records:

The first and third rows represent the same logical event. But their delivery identifiers are different. The patch checks whether a delivery_id has already been seen. Conceptually:

seen_delivery_ids = set()

for record in deliveries:
    delivery_id = record["delivery_id"]

    if delivery_id in seen_delivery_ids:
        continue

    seen_delivery_ids.add(delivery_id)
    process(record)

Let’s trace the behaviour.
First record: del-001 The identifier hasn’t been seen before, so the record is processed.
Second record: del-002 This identifier is also new, so the record is processed.
Third record: del-003 Although it represents the same event as the first record, its delivery_id is different.

The identifier hasn’t been seen before, so the record is processed again. Nothing is removed.

Why Does the Patch Fail?

Because delivery_id identifies the arrival, not the underlying business event. The source guarantees that every arrival receives a new delivery identifier. That includes retries. Therefore, deduplicating on delivery_id doesn’t detect repeated deliveries of the same event. The patch is performing exactly the operation it was programmed to perform. It’s simply using the wrong identity.

The team introduces a deduplication step that keeps the first delivery for each delivery_id. However, the source assigns a fresh delivery_id to every arrival, including redeliveries of an existing event. Before the patch, the revenue total is: 79,641,107 cents

We’ve Seen This Mistake Before

In F04.2, the team tried to prevent duplicate database writes using an idempotency key containing the task attempt ID. The key was: (job, task, attempt, customer)

That implementation failed because each retry received a new attempt ID. The sink treated every retry as a new logical operation. The problem here is almost identical. In F04.2, the team identified the attempt instead of the logical write. Here, the team identifies the delivery instead of the logical event. In both cases, the deduplication mechanism works, but the chosen identifier doesn’t represent what needs to be deduplicated.

How Should We Think About Event Identity?

Before implementing deduplication, we need to answer: What exactly counts as the same event?
For our source:

• Two records with the same event_id represent deliveries of the same logical event.
• Two records with different delivery_id values may still represent that same event.
• A correction has its own event_id, but references the earlier event it replaces through supersedes_event_id.

This gives us two distinct responsibilities:

1. 01Recognise repeated deliveries of the same event.
2. 02Apply correction events according to the business contract.

Deduplicating by event_id addresses the first responsibility. However, it doesn’t automatically handle the second. A correction is a new event with its own identity, even though it replaces an earlier event. We need to account for both relationships to calculate the correct revenue.

We’ll explore this in the next section.

In the previous section, we discovered why deduplicating by delivery_id didn’t solve our revenue problem. Every arrival receives a new delivery identifier, even when the source is delivering the same event again. The team now understands the mistake. Instead of identifying individual deliveries, we need to identify the logical events they represent. So the engineers make another change. They replace delivery_id with event_id in the deduplication logic.
This time, when the same event arrives more than once, the pipeline keeps only its first delivery. That should eliminate duplicate order amounts. But will it produce the correct revenue total?

Step 1: Deduplicate by Event ID

Consider our earlier example:

When deduplicating by event_id, the pipeline processes:

• evt-101 from the first delivery.
• evt-102 from the second delivery.

It discards the third delivery because evt-101 has already been processed. This correctly handles redelivery. However, our feed contains another type of record: correction events. A correction has a new event_id, even though it replaces an earlier event.

Let’s examine why this matters.

Step 2: What Happens to Correction Events?

Suppose an order was originally recorded with an amount of 1,000 cents. Later, the order management system discovers that the correct amount was 1,200 cents. It produces a correction event.

Here:

• evt-201 is the original event.
• evt-202 is a correction event.
• evt-202 replaces evt-201.

Under our business contract, the correct revenue contribution is: 1,200 cents

But if we simply deduplicate by event_id and sum the remaining events, both records are retained because they have different event IDs. The pipeline calculates: 1,000 + 1,200 = 2,200 cents

The duplicate deliveries are gone, but the original event and its correction are still counted together. The result remains incorrect.

Our feed contains:

• Repeated deliveries of some events.
• 44 correction events that replace earlier events.
• A correct revenue total of 74,558,550 cents.

Why Is This Particularly Dangerous?

The original error was approximately 6.8%, which was large enough to attract attention. After the patch, the error is much smaller. The team might conclude that the problem has been fixed because the total is now close to the expected result. But a smaller error is not the same as a correct result. The patch fixed repeated deliveries without handling event corrections.

We need a processing state that understands both.

Step 3: Build a State That Understands Events and Corrections

Instead of maintaining only a running sum, we can maintain an event ledger. The ledger stores two pieces of information:

1. 01Event entries: A set of (event_id, amount) pairs.
2. 02Superseded event IDs: A set of event IDs that have been replaced by correction events.

For example, consider these records:

The second row is a redelivery of evt-101. The third row is a correction that replaces evt-101. The fourth row is an unrelated event. Our ledger state becomes:
Event entries

{
  (evt-101, 1000),
  (evt-102, 1200),
  (evt-103, 500)
}

Notice that evt-101 appears only once in the set, even though it was delivered twice.

Superseded event IDs

{
  evt-101
}

This tells us that evt-101 has been replaced by another event.

Step 4: Calculate Revenue from the Ledger

To calculate the final revenue:

1. 01Examine the event entries.
2. 02Exclude entries whose event_id appears in the superseded set.
3. 03Sum the amounts of the remaining events.

For our example:

The correct revenue is: 1,200 + 500 = 1,700 cents

The repeated delivery is counted only once, and the original event is excluded because a correction replaced it. This handles both problems.

Step 5: Can This State Work Across Multiple Map Tasks?

So far, we’ve described the ledger within one process. But our real job uses multiple map tasks, a shuffle, and reduce tasks. The same event may be delivered to different map partitions.
For example:

• Map Task 1 receives the first delivery of evt-101.
• Map Task 7 receives another delivery of evt-101.
• Map Task 9 receives its correction, evt-102.

We need to combine the partial states without reintroducing duplicates. This is where the mergeable-state concept from F03.1 becomes important. Each map task builds its own event ledger. The reduce stage combines these ledgers using set union. For example:

Map Task 1
Entries: {(evt-101, 1000)}

Map Task 7
Entries: {(evt-101, 1000)}

Map Task 9
Entries: {(evt-102, 1200)}
Superseded: {evt-101}

After merging:

Entries:
{
  (evt-101, 1000),
  (evt-102, 1200)
}

Superseded:
{
  evt-101
}

The final revenue contribution is 1,200 cents.

Why Does Set Union Work?

Set union has two useful properties for distributed processing.

1. It is idempotent.
Adding an item that already exists doesn’t change the set. For example:

{evt-101} ∪ {evt-101} = {evt-101}

Therefore, receiving the same event twice doesn’t create another copy in the merged state.

2. It is mergeable.
Partial sets can be combined across tasks without depending on the order in which they are merged. For example:

(A ∪ B) ∪ C = A ∪ (B ∪ C)

This allows map tasks to build partial ledgers independently and reduce tasks to merge those states. The final calculation is performed after the event entries and supersession information have been combined. This matters because a correction and the event it replaces may arrive in different partitions.

What Does This Cost?

The ledger requires more intermediate state than a simple sum. A sum only needs a customer key and a running total. Our simulator models that as:

• Customer ID: 10 bytes.
• Revenue total: 8 bytes.

18 bytes per customer state. The event ledger stores more information:

• Each event entry requires approximately 20 bytes.
• Each superseded event ID requires another 10 bytes.

As a result, the shuffle volume increases.

The ledger requires approximately 2.4 times more intermediate data. However, it produces the correct results. In our simulator, the total runtime remains almost unchanged:

• Sum aggregation: 204.8 seconds.
• Ledger aggregation: 204.9 seconds.

These runtimes are specific to the model and workload. Real systems may experience additional memory, serialization, or shuffle overhead from maintaining the larger state.

What Should We Learn?

The smallest intermediate state isn’t always the correct one. A simple sum is compact and efficient, but it doesn’t preserve enough information to distinguish new events, redeliveries, and corrections. The ledger stores additional information because the business contract requires it.

We’ve now developed a processing state that handles repeated deliveries and corrections. Our simulator produces the correct total: 74,558,550 cents across 796 customers.
But how do we know the pipeline will remain correct when events arrive in different partitions or workers fail? A successful run on one dataset isn’t enough. We need tests that deliberately reproduce the conditions that caused the original failures. For this lesson, we’ll use three types of tests:

1. 01Truth test: Compare the pipeline against an independent implementation of the business contract.
2. 02Failure-injection test: Introduce failures and verify that the published results remain correct.
3. 03Determinism test: Run the pipeline repeatedly under different execution conditions and verify that the results don’t change.

We’ll cover the first two here. The third appears in Section 07.

Test 1: Compare Against the Contract Truth

Before testing the distributed pipeline, we need an independent reference implementation. This is sometimes called a test oracle. For our nightly ledger, the oracle calculates revenue in a single process using the business rules. The contract says:

• A delivery belongs to the current run only if it satisfies the fixed reporting cutoff.
• Multiple deliveries with the same event_id represent one logical event.
• A correction event replaces the earlier event named by supersedes_event_id.
• The final result contains the revenue of events that haven’t been superseded.

The reference implementation applies these rules directly. It produces: 74,558,550 cents across 796 customers.
This becomes our contract truth for the fixture.

Why Compare Customer by Customer?

Checking only the grand total isn’t sufficient. Two incorrect customer totals could offset each other and produce the correct overall revenue. For example:

The overall total matches, but both customer records are wrong. Therefore, our truth test compares the expected and actual results key by key. For every customer, it verifies the revenue amount, and it also checks that no expected customers are missing and no unexpected customers are present.

What Does the Truth Test Reveal?

Let’s run the test against all four processing strategies.

The test clearly distinguishes a partial improvement from a correct implementation. The event-ID patch reduces the error significantly, but it still fails because 41 customers have incorrect totals. Only the ledger implementation passes all customer-level comparisons.

Why Was Last Month’s Dataset Not Enough?

Recall that the previous fixture contained:

• 21,154 deliveries.
• No redeliveries.
• No corrections.

All four processing strategies produce the same correct total on that fixture: 75,387,030 cents
That test would pass even with the unsafe sum implementation. It cannot detect a bug involving duplicate deliveries or corrections because the fixture contains neither. This teaches an important testing principle:
A test must contain the conditions that can expose the behavior we’re trying to verify.

A useful regression test should fail on the original unsafe implementation and pass after the defect is corrected.

Test 2: Inject Failures and Verify the Published Results

The truth test checks whether the aggregation correctly interprets the source events. But it doesn’t prove that the results will be published safely when tasks fail. Remember what we learned in F04.2. A publish task may run more than once because of:

• Lost completion acknowledgments.
• Worker failures.
• Zombie attempts.
• Speculative execution.

These situations can duplicate external writes even when the aggregation itself is correct. To test this behaviour, we deliberately introduce failures into the execution. This is called failure injection.

How Does Failure Injection Work?

Our simulator lets us specify a particular failure scenario. For example:

• Lose the completion acknowledgement for Publish Task 3.
• Disconnect Publish Task 5 from the coordinator while it continues running.
• Kill Worker 4 halfway through Publish Task 4.

The pipeline executes with the injected failure. Afterwards, the test verifies the final published results. For our ledger, the checks include:

1. 01The ledger contains one row per expected customer.
2. 02Customer revenue totals match the contract truth.
3. 03The modelled notification system produces one email per customer.
4. 04The failure doesn’t leave unexpected duplicates or missing results.

Compare an Unsafe and a Retry-Safe Configuration

Let’s test two implementations.
Configuration A: Unsafe

• Append-only ledger sink.
• Emails sent directly from publish tasks.

Configuration B: Retry-safe under the simulator’s rules

• Atomic ledger sink with one commit per logical task.
• Notifications recorded in a transactional outbox.
• A controlled outbox sender that processes each committed notification once.

Both configurations perform the same revenue computation. The difference is how they handle repeated external side effects. Our simulator produces the following results:

What Do These Results Tell Us?

The unsafe implementation fails all three scenarios.
Lost completion message
The first attempt writes successfully but isn’t acknowledged. The retry writes the same 78 customer records again. The append-only sink keeps both copies.

Zombie task
The coordinator retries Publish Task 5 even though its original attempt continues running. Both attempts write the same 59 customer records and send notifications.

Worker failure during publishing
Publish Task 4 writes 36 customer records before its worker fails. The retry writes the complete set again, leaving the first 36 records duplicated. The atomic sink and outbox configuration avoids these duplicate committed results under the simulator’s assumptions.
It maintains:

• 796 customer records
• Correct revenue totals
• No duplicate notifications in the tested scenarios

What Does This Prove?

It demonstrates that the safe configuration preserves the expected results under the specific failures we injected. The important idea is that we test the failure behaviour rather than assuming our recovery mechanisms are correct. However, these tests don’t establish correctness for every possible real-world failure. For example, the simulator doesn’t test what happens when an outbox sender successfully delivers an email but crashes before recording delivery progress. A production test suite would need to cover that failure separately.

Why Test the Unsafe Version Too?

Suppose we write a test that confirms our new implementation produces 796 ledger rows. That’s useful, but it doesn’t tell us whether the test would detect the original bug. A stronger approach is to run the same test against the unsafe implementation. If the unsafe version fails and the corrected version passes, the test demonstrates that it can catch the regression we care about.

For our example:

• The append-only implementation fails under lost acknowledgments.
• The atomic implementation passes the corresponding ledger correctness checks.
• The direct-email implementation produces duplicates.
• The modelled outbox implementation avoids them.

The test doesn’t merely confirm that the job completed. It verifies the intended behaviour when execution doesn’t go according to plan.

Let’s return to the event ledger itself. Our processing state contains:

• A set of (event_id, amount) pairs.
• A set of superseded event IDs.

Does This Still Work When a Worker Fails?

Our simulator tests the event ledger in three configurations.

All three executions produce the same customer revenue results. This demonstrates two important properties of the ledger state.

1. Duplicate event deliveries are safe under the modeled merge rules.
If the same event appears in multiple map tasks, merging their sets doesn’t count the event twice.

2. Recomputing map tasks doesn’t change the logical result.
If a worker fails and a map task is repeated, the same event entries may be produced again. The merged state still represents the same distinct events and supersession relationships. These properties make the event ledger suitable for retryable distributed aggregation under our data contract.

What Is the Third Test?

We’ve now covered:

• The truth test, which checks business correctness against an independent reference.
• The failure-injection test, which checks recovery and publishing behavior under selected failures.

The third test is the determinism test. It verifies that different execution schedules and retries still produce the same final result. We’ll explore that in Section 07.

Our experiments demonstrate three important lessons about building and testing retry-safe data pipelines.

1. At-Least-Once Delivery Requires More Than a Simple Sum

Our new order feed contains 22,363 deliveries of 21,198 distinct events. Some events arrive multiple times, and others are corrections that replace earlier events. The original revenue aggregation simply adds the amounts from every qualifying delivery.

As a result, it produces 79,641,107 cents instead of 74,558,550 cents
The calculation overstates revenue by 6.82%. The aggregation is mathematically valid for the records it receives, but it doesn’t implement the business contract.

2. Deduplication Must Use the Correct Identity

Deduplicating by delivery_id removes nothing because every delivery has a unique identifier. Deduplicating by event_id removes redeliveries, but correction events still require special handling. The event ledger addresses both requirements by maintaining:

• The set of distinct event entries.
• The set of event IDs that have been superseded.

It then calculates revenue using only the event entries that haven’t been replaced.

3. Correctness Must Be Tested Under Failures

The ledger state produces the correct result in a single process, across distributed tasks, and under the tested worker failure. The truth test catches incorrect customer totals. Failure-injection tests catch duplicate external writes caused by retries. The determinism test, which we’ll examine next, checks whether repeated executions produce identical results. Together, these tests provide evidence that the pipeline preserves its intended behaviour under the tested scenarios.

Limitations of the Model

Our simulator makes several simplifying assumptions.

• Correction semantics are defined by the contract: In this lesson, a correction replaces the earlier event named by supersedes_event_id. Other systems may use corrections that adjust an existing amount rather than replace it. Those contracts require different processing rules.
• Late arrivals follow a defined window policy: The current run includes only records admitted by its fixed reporting cutoff. Later arrivals belong to a subsequent processing or correction process rather than silently changing the completed run.
• One injected failure per run: Our failure-injection tests examine individual scenarios. Real systems can experience simultaneous worker failures, repeated infrastructure problems, or coordinator failures.
• The coordinator remains available: The simulator assumes the coordinator survives and can schedule retries. Production recovery may require additional mechanisms when the coordinator itself fails.
• The event ledger maintains more state: Tracking event IDs and superseded IDs requires more memory and shuffle data than a simple running sum. Production implementations also need to plan for long-term state growth and retention.
• The fixture represents one workload: Its redelivery frequency, correction patterns, and late-arrival distribution may differ from real production data. Performance and correctness tests should include additional representative cases.
• External delivery guarantees are simplified: The simulator’s atomic sink and outbox tests cover selected publishing failures. They don’t establish exactly-once delivery for every possible failure involving an external email service.

What This Means for Production Systems

A distributed data pipeline may successfully process all its records and complete every task while still producing incorrect business results.

For example:

• Redelivered events can cause duplicate revenue contributions.
• Corrections can be counted alongside the records they replace.
• Publish retries can duplicate external database writes.
• Time-dependent processing can cause the same reporting window to produce different results.

These problems occur at different points in the pipeline. Therefore, no single optimisation or retry mechanism solves all of them. We need to define the expected behaviour at each boundary:

1. 01Source boundary: How are repeated deliveries and corrections interpreted?
2. 02Processing boundary: Can partial states be merged and recomputed without changing the intended result?
3. 03Sink boundary: Can repeated task attempts publish data without creating duplicates or partial results?
4. 04Testing boundary: Can we reproduce the failures and verify that the final output remains correct?

We’ll explore source acknowledgements, replay, and delivery semantics further in the Kafka curriculum, and task recovery and output commits in the advanced Spark curriculum.

In the previous sections, we built an event ledger that correctly handles repeated deliveries and correction events. We also introduced two tests:

1. 01Truth test: Verify that the pipeline produces the correct customer revenue according to the business contract.
2. 02Failure-injection test: Introduce failures and verify that retries don’t produce duplicate or missing results.

Now let’s examine the third test: determinism testing. This test answers an important question:
Will our pipeline produce the same results when tasks execute at different times, on different workers, or after failures?

Why Do We Need a Determinism Test?

Recall what we learned in F04.1. Our nightly ledger processes events belonging to a defined reporting window. However, some events arrive after the window closes. We previously compared two approaches to selecting records.

Approach 1: Fixed reporting cutoff
Every task receives the same reporting cutoff as a job parameter. For example:

if record["ingest_time_utc"] <= cutoff_time:
    process(record)

The cutoff remains unchanged throughout the job, including retries.

Approach 2: Current system time
Each task decides which records to process based on the time at which it executes. For example:

from datetime import datetime, timezone

if record["ingest_time_utc"] <= datetime.now(timezone.utc):
    process(record)

This approach creates a problem. A task that retries later may see additional events that weren’t available during its first attempt. As a result, two executions of the same reporting job can produce different revenue totals.

What Should Deterministic Processing Guarantee?

For our nightly ledger, we expect the following: Given the same logical input, reporting cutoff, and processing rules, the pipeline should produce the same customer-level results regardless of task scheduling or retries.

For example, these executions should produce the same result:

• A normal execution without failures.
• An execution where a map task fails and retries.
• An execution where a reduce task must recompute lost intermediate results.
• An execution where a task’s completion acknowledgement is lost.
• A rerun that starts later but uses the same reporting cutoff and logical input.

The scheduler may assign tasks to different workers. Some tasks may execute more than once. The total runtime may change. But the final customer revenue should remain the same.

How Do We Compare Results Across Runs?

We could compare every customer record manually, but that becomes inconvenient when the ledger contains thousands or millions of records. Instead, we can compute an output digest. A digest is a compact fingerprint of the complete result, calculated using a hash function. For example, suppose our ledger contains:

We can:

1. 01Sort the records by customer ID.
2. 02Serialize their keys and values in a consistent format.
3. 03Calculate a hash of the resulting data.

Conceptually:

import hashlib

def calculate_digest(customer_revenue):
    rows = sorted(customer_revenue.items())

    content = "\n".join(
        f"{customer_id}:{revenue}"
        for customer_id, revenue in rows
    )

    return hashlib.sha256(content.encode()).hexdigest()

If two executions produce exactly the same customer records, they should produce the same digest. Sorting matters because distributed systems don’t always return records in the same order. Two result sets can contain identical data but arrive in different orders. Canonicalising the result before hashing prevents row ordering from causing false differences.

Our nightly ledger uses a fixed reporting cutoff. The source data and processing rules remain the same. We execute the job six times:

• Once without failures.
• Several times with different injected worker or task failures.
• Once starting later than usual.

Why Does the Fixed-Cutoff Version Produce One Digest?

Every task uses the same reporting cutoff, even when the job starts later or a task retries. The event ledger also handles repeated deliveries and corrections using stable event identities. Therefore, the execution schedule doesn’t change the logical result. The map tasks may execute in a different order. A worker may fail and force re-computation. The reduce tasks may merge intermediate states in a different sequence. But the final customer-level result remains unchanged.
All six executions produce: 74,558,550 cents across 796 customers.
Since the complete result is identical, the simulator produces one distinct digest.

Why Does the Clock-Based Version Produce Five Digests?

In the clock-based implementation, tasks select records according to the time at which they execute. If a worker fails, its retry begins later. By then, additional late-arriving order events may have become available. The retry can therefore include events that the original attempt didn’t process. For example:

• Normal execution produces 75,090,421 cents.
• A failure during Map Task 7 produces 75,117,419 cents.
• A failure during Map Task 2 produces 75,150,666 cents.
• Starting the job seven minutes after the window closes produces 75,371,869 cents.

The six runs produce five distinct results. Two happen to match, but the others don’t. The output depends on when particular tasks execute. This violates our requirement that all runs of the same reporting window produce the same logical result.

Does One Digest Prove That a Pipeline Is Correct?

Not by itself. A pipeline could consistently produce the same incorrect result. For example, our original sum aggregation could repeatedly calculate 79,641,107 cents for the same duplicate-containing input. Every run might produce the same digest, yet the result would still violate the business contract. This is why we need all three tests.

Each test checks a different aspect of correctness. Together, they provide stronger evidence than any one test alone.

How can we recognise when a distributed pipeline isn’t handling redeliveries, corrections, or failures correctly? Our simulator provides several useful warning signs.

Warning Sign 1: The Revenue Changes After a Source Upgrade

Suppose the revenue pipeline has been running successfully for months. Then the source introduces at-least-once delivery. Without changing the aggregation code, the reported revenue increases by nearly 7%. That should raise an important question:

Did the business generate additional revenue, or did the source deliver some events more than once?
In our fixture, the correct revenue is 74,558,550 cents, but summing every delivery produces 79,641,107 cents.

The source changed its delivery behavior, not the underlying business activity represented by the contract.

Warning Sign 2: Deduplication Doesn’t Change the Result

The team adds deduplication using delivery_id. However, the revenue total remains unchanged. This suggests that the deduplication identifier doesn’t represent the logical events being repeated. Since the source assigns a new delivery ID to every arrival, no repeated IDs are found. The deduplication operation executes successfully but removes nothing.

Warning Sign 3: A Fix Reduces the Error Without Eliminating It

Deduplicating by event_id removes repeated event deliveries. The revenue overstatement decreases from 5,082,557 to 946,893 cents. However, 41 customers still have incorrect totals. This indicates that another part of the data contract remains unhandled. In our case, correction events are still being counted alongside the earlier events they replace. A smaller error isn’t evidence that the pipeline is correct.

Warning Sign 4: The Same Reporting Window Produces Different Results

Suppose a job generates different customer totals depending on when it starts or whether a worker fails. For example:

• Normal run: 75,090,421 cents.
• Run with a worker failure: 75,117,419 cents.
• Later rerun: 75,371,869 cents.

This suggests that task execution time or changing external input is influencing the result. If the reporting contract requires a fixed snapshot, investigate whether the job is using stable input boundaries and deterministic processing rules.

Warning Sign 5: Tests Pass Only on Clean Data

A test suite may pass when the input contains:

• No duplicate deliveries.
• No correction events.
• No late-arriving records.
• No task failures.
• No repeated external writes.

But those tests don’t verify how the pipeline behaves when these conditions occur. For example, our previous fixture contains no redeliveries or corrections. All four aggregation strategies produce the correct result on that fixture. The defective implementations are exposed only when the new fixture includes the behaviours they fail to handle.

What Should You Investigate?

When diagnosing retry-safety problems, examine:

• Whether the source distinguishes event identity from delivery identity.
• How many events were delivered more than once.
• Whether correction events are applied according to the business contract.
• Whether the processing state supports safe merging and repeated input.
• Whether different execution schedules produce the same customer-level results.
• Whether task retries cause repeated external writes.

Before trusting a pipeline that receives data from an at-least-once source, ask five questions.

1. What Identifies the Logical Event?

Determine which identifier represents the business event rather than an individual delivery attempt. For our order feed:

• event_id identifies the event.
• delivery_id identifies one arrival.
• supersedes_event_id identifies the event replaced by a correction.

Use the appropriate identity for each operation.

2. Is the Aggregation State Safe Under Redelivery?

Ask what happens when the same event arrives multiple times. A running sum adds its amount every time. A set keyed by event identity can recognise that the event is already represented. However, event identity must be enforced consistently. If the same event ID arrives with conflicting values, the pipeline needs an explicit conflict-resolution rule rather than silently treating them as independent events.

3. Can Intermediate States Be Merged Correctly?

Distributed processing requires partial states to be combined across workers. For our event ledger, merging involves:

• Taking the union of event entries.
• Taking the union of superseded event IDs.
• Calculating the final total after the relevant information has been combined.

The merge operation should produce the correct result regardless of how records are partitioned or how partial states are grouped.

4. How Does the Business Contract Handle Corrections?

A correction isn’t necessarily a duplicate. In our example, a correction replaces the event it names. Other systems may represent corrections as adjustments that should be added to the existing amount. The processing logic must implement the actual business contract.

5. Which Failure Scenarios Have We Tested?

A pipeline should be tested under realistic execution failures. At minimum, consider:

• Duplicate source deliveries.
• Lost task completion acknowledgements.
• Worker failures during processing.
• Worker failures after partial external writes.
• Different task execution schedules.
• Reruns using the same reporting cutoff.

Verify both the final customer values and the behaviour of external sinks.

Let’s bring together what we’ve learned.

One More Question

Suppose we run our truth test using last month’s input fixture. That fixture contains:

• 21,154 deliveries
• 21,154 distinct events
• No redeliveries
• No corrections

We compare four processing strategies:

1. 01Sum every delivery.
2. 02Deduplicate by delivery_id.
3. 03Deduplicate by event_id.
4. 04Maintain an event ledger that handles corrections.

Will they produce different revenue totals?

Try It Yourself

When the L-F04 workbench becomes available, you’ll be able to run the same tests against the simulated coordinator.

Experiment 1: Test Determinism Using the Current Clock

Run:

The simulator executes the job six times under different execution conditions.

Expected result: Six runs produce five distinct output digests.

This shows that the output changes when tasks select input records using the current clock.

Now run:

This configuration uses a fixed reporting cutoff.
Expected result: Six runs produce one distinct output digest.

The executions may follow different schedules, but the final result remains the same.

Experiment 2: Compare Aggregations Against the Contract Truth

Run the truth test using the original sum aggregation:

Expected results:

• Revenue: 79,641,107 cents.
• Customers with incorrect totals: 164.

Now test the event ledger:

Expected results:

• Revenue: 74,558,550 cents.
• Customers with incorrect totals: 0.

The ledger passes because it correctly handles repeated deliveries and correction events.

What Should You Observe?

The two experiments verify different properties.
The truth test checks whether the calculation follows the business contract. The determinism test checks whether the result remains consistent under different execution conditions. A pipeline must satisfy both requirements.

Note: The L-F04 workbench is planned. These commands describe its expected behaviour once available.

An at-least-once source may deliver the same business event multiple times. If a distributed aggregation treats every delivery as a new event, it can produce incorrect totals even when all its tasks execute successfully.

In this lesson, we learned that:

1. 01Events and deliveries are different: One logical event may have multiple deliveries, each with a different delivery ID.
2. 02A simple sum isn’t always retry-safe: Adding every delivery can count the same business event multiple times.
3. 03Deduplication requires the correct identity: Deduplicating by delivery_id doesn’t remove event redeliveries when each arrival receives a new ID.
4. 04Corrections require explicit handling: Deduplicating by event_id removes redeliveries but doesn’t automatically remove superseded event contributions.
5. 05Mergeable event state supports distributed recovery: A ledger containing event entries and superseded IDs can be combined across workers while preserving the intended result.
6. 06Correctness must be demonstrated through tests: Truth tests, failure injection, and determinism tests verify different parts of the pipeline’s contract.

Our Revenue Example

The original sum was 6.82% too high. Deduplicating by event ID removed the repeated deliveries but left an overstatement of 946,893 cents because corrections weren’t properly applied. The event ledger produced the correct result by tracking both distinct events and superseded IDs.

The Three Tests That Establish Confidence

Together, these tests verify that the pipeline interprets source data correctly, handles the selected recovery scenarios, and reproduces the same results across repeated executions. They provide strong evidence for the tested cases, rather than a universal guarantee against every possible failure.