Retry and Alerting

All the runs in the previous lesson had one thing in common: none of them failed. The task either ran or was never triggered; when it ran, it wrote all of the rows it saw. The partial run was harmless there, because the write was keyed and the missing part was completed in the next run. This lesson removes that guarantee.

In a real pipeline there is a third outcome: the task runs, starts writing and fails in the middle. What is left behind is neither an empty partition nor a complete one. This lesson measures two questions — what a half-written partition does to serving, and whether retrying the failed task repairs the pipeline or duplicates rows. The answer to the first question is the strongest example of the third claim of the course.

• OO17. The source, the pipeline and the task run are fictional; no real orchestrator, store or alerting service is called. The seed is 20260218, and the measurement does not depend on the run.
• OO18. An atomic write means that a task either writes all of its rows or leaves none behind. In a write that is not atomic, the rows left halfway stay readable.
• OO19. The task fails at row 120 while writing the 2026-07 partition. This number is a choice of place; what is measured is not the number itself but the difference between half and nothing.
• OO20. The retry, backoff, circuit breaker and graceful degradation procedure was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and in the Resilience and Reliability course of the System Design and Distributed Systems curriculum and is not repeated. The only question here is whether a retry duplicates rows.
• OO21. The term for being safe to repeat is idempotent. Reprocessing duplicating rows was measured in the Introduction to Data Engineering course; the measurement here shows the same fact at task level.
• OO22. Two alert sources are separated: the task alert goes out if the task is still failing after the last try, the data alert comes from the checks.
• OO23. The baseline is the run without orchestration: no atomic write, no retry, no alert.
• OO24. The counted units: rows written, duplicated rows, consumption rows, serving deviation, tries. No raw duration is written. No ratio is printed on serving, the value itself is written.

Failing in the Middle of a Task

The task that writes a partition is not a single operation; it writes the rows in order and can fail after any row. What it leaves behind depends on whether the write is atomic. In an atomic write the half-finished work is never visible — the task fails, the partition stays empty, and the pipeline reads that period as if it did not exist. In a write that is not atomic, the first 120 rows stay where they are, and as they pass through the pipeline they cannot be told apart from complete rows.

The measurement is made with two checks. The row count check looks at whether the raw row count deviates from the expected one by more than five percent. The period-aware completeness check looks at whether every pair of region and period exists in the data; the Introduction to Data Engineering course had separated this check from its flawed sibling, and the correct one is used here.

# FICTIONAL. The meter readings, the five-stage pipeline and the task run are
# defined inside the lesson; no real store, orchestrator or alerting service is called.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


REGION = ["east", "west", "north", "south", "central"]
TYPE = ["residential", "commercial"]
PERIODS = [f"2026-{a:02d}" for a in range(1, 13)]
SUBSCRIBERS = {}
for i in range(200):
    r = rng(SEED + 37 * i)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "start": 10000 + int(r() * 60000)}


def generate_source():
    records = []
    for i, (sub, a) in enumerate(SUBSCRIBERS.items()):
        e = a["start"]
        for p in PERIODS:
            r = rng(SEED + 101 * i + 7 * PERIODS.index(p))
            e += 4 + int(r() * 46)
            records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"],
                            "index": e, "unit": "m3"})
    return records


def s1_screen(records):
    REQUIRED = ("subscriber", "period", "region", "index", "unit")
    keep = [k for k in records if all(a in k for a in REQUIRED)]
    return keep, len(records) - len(keep)


def s2_dedupe(records):
    seen, keep = set(), []
    for k in records:
        key = (k["subscriber"], k["period"])
        if key in seen:
            continue
        seen.add(key)
        keep.append(k)
    return keep, len(records) - len(keep)


def s3_consumption(records):
    table = {(k["subscriber"], k["period"]): k for k in records}
    out, dropped = [], 0
    for k in records:
        i = PERIODS.index(k["period"])
        if i == 0:
            continue
        prev = table.get((k["subscriber"], PERIODS[i - 1]))
        if prev is None:
            dropped += 1
            continue
        try:
            f = int(k["index"]) - int(prev["index"])
        except (TypeError, ValueError):
            dropped += 1
            continue
        out.append(dict(k, m3=f))
    return out, dropped


def s4_summary(records):
    totals = {}
    for k in records:
        key = (k["region"], k["period"])
        totals[key] = totals.get(key, 0) + k["m3"]
    return totals, 0


def s5_serve(summary):
    region = {}
    for (b, d), v in summary.items():
        region.setdefault(b, []).append(v)
    return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0


def read_result(records):
    v, _ = s1_screen(records)
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    summ, _ = s4_summary(v)
    serving, _ = s5_serve(summ)
    return serving, len(v)


def write_partition(records, period, cutoff=None, atomic=False):
    """cutoff: the task fails after this many rows. atomic: a half-written partition is not visible."""
    target = [k for k in records if k["period"] == period]
    if cutoff is None or cutoff >= len(target):
        return target, False
    partial = target[:cutoff]
    return ([], True) if atomic else (partial, True)


def check_row_count(data, expected=2400, margin=0.05):
    return abs(len(data) - expected) / expected > margin


def check_period_aware_completeness(data):
    """Every region exists in every period."""
    seen = set((k["region"], k["period"]) for k in data)
    return seen != set((b, d) for b in REGION for d in PERIODS)

Half Versus Nothing

The second block runs the same failure in two write modes and puts four things side by side: the rows written to the partition, the raw rows of the store, the consumption rows and the deviation of serving from the baseline.

SOURCE = generate_source()
BASELINE_SERVING, BASELINE_N = read_result(SOURCE)
OTHERS = [k for k in SOURCE if k["period"] != "2026-07"]
print("baseline consumption:", BASELINE_N)
print("baseline serving:", BASELINE_SERVING)
print(f"{'setup':<16}{'written':>8}{'raw':>6}{'consumption':>13}{'row check':>11}{'completeness':>14}")
for name, atomic in (("not atomic", False), ("atomic", True)):
    written, failed = write_partition(SOURCE, "2026-07", 120, atomic)
    store = OTHERS + written
    serving, n = read_result(store)
    print(f"{name:<16}{len(written):>8}{len(store):>6}{n:>13}"
          f"{str(check_row_count(store)):>11}{str(check_period_aware_completeness(store)):>14}")
    print("   deviation:", {b: round(serving[b] - BASELINE_SERVING[b], 2) for b in BASELINE_SERVING})

baseline consumption: 2200
baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55}
setup            written   raw  consumption  row check  completeness
not atomic           120  2320         2040      False         False
   deviation: {'central': -92.63, 'north': -121.82, 'east': -49.09, 'west': -49.27, 'south': -61.55}
atomic                 0  2200         1800       True          True
   deviation: {'central': -2.89, 'north': 6.12, 'east': -3.25, 'west': 12.8, 'south': 7.01}

The first row is the strongest example of the third claim of the course. The half-written partition makes serving wrong in all five regions: the deviation is between −49.09 and −121.82, and none of it is small. In return, both checks are green. The raw rows are 2320, which deviates from the expected 2400 by 3.33% and stays inside the band of 5%. The period-aware completeness check passes too, because the 120 rows written carry subscribers from all five regions — the 2026-07 period exists in the data for every region, it is only incomplete. The number looks reasonable, serving is wrong, and no check says so.

Narrowing the band is the first fix that comes to mind, and it brings its own cost. To see this half-write, the band has to go below 3.33%. In the previous lesson four subscribers ended their subscriptions, which means 16 rows over four periods, that is 0.67% of the raw rows; for the band not to fire on that change, it has to stay above 0.67%. The threshold can be tuned, but only between 0.67% and 3.33%; choosing within that range requires knowing in advance both the size of the error to be caught and the size of the movement to be ignored.

The second row looks worse at first glance. In the atomic write the partition is never written; the store drops to 2200 rows and consumption falls to 1800 instead of 2040, that is, 240 rows fewer are computed than in the half-write. The reason the loss grows is the graph of the earlier lesson: if 2026-07 does not exist at all, the consumption of 2026-08 cannot be computed either, because there is no index of the previous period.

Looking at serving, though, the order reverses. The setup that loses more rows has a deviation between −3.25 and +12.80; the one that loses fewer falls as far as −121.82. The reason is this: serving is the period average per region. When the partition is never written, the cell of that period does not enter the average at all, and the remaining eleven periods give the correct average. When it is written halfway, the cell enters the average, but it holds one hundred and twenty subscribers instead of two hundred — a low number joins the average as if it were a correct one.

Moreover, in the atomic setup both checks fire. The raw rows are 2200 and exceed the band of five percent; and because the 2026-07 period is absent from the data for every region, the period-aware completeness check sees the corruption too. The sentence that comes out of this is the backbone of this lesson: atomicity does not make the answer correct; it makes the failure visible and bounded. The result is still incomplete and the deviation is still not zero; what is gained is that the incompleteness can be caught and its effect does not go beyond one period.

Retry and Alerting

The failed task is retried. The third block tries the task three times — the first two tries fail again at row 120, the third completes — and counts what it changes whether the write appends or is keyed. The task alert column carries the task alert: it fires if the task is still failing after the last try.

def task_run(records, period, cutoffs, atomic=False, keyed=False):
    """The task is tried in order; cutoffs give the row at which each attempt fails.
    None = the attempt succeeds. The write either appends or is keyed."""
    store, written, tries, failed = [k for k in records if k["period"] != period], [], 0, False
    for cutoff in cutoffs:
        tries += 1
        part, failed = write_partition(records, period, cutoff, atomic)
        if keyed:
            table = {(k["subscriber"], k["period"]): k for k in written}
            for k in part:
                table[(k["subscriber"], k["period"])] = k
            written = list(table.values())
        else:
            written = written + part
        if not failed:
            break
    return store + written, tries, len(written), failed


print(f"{'setup':<24}{'tries':>7}{'written':>8}{'raw':>6}{'consumption':>13}{'north':>9}"
      f"{'task alert':>12}{'row check':>11}{'completeness':>14}")
for name, cutoffs, atomic, keyed in (
        ("no retry , not atomic", (120,), False, False),
        ("no retry , atomic", (120,), True, False),
        ("three tries , appending", (120, 120, None), False, False),
        ("three tries , keyed", (120, 120, None), False, True)):
    store, tries, written, failed = task_run(SOURCE, "2026-07", cutoffs, atomic, keyed)
    serving, n = read_result(store)
    print(f"{name:<24}{tries:>7}{written:>8}{len(store):>6}{n:>13}"
          f"{round(serving['north'] - BASELINE_SERVING['north'], 2):>9}{str(failed):>12}"
          f"{str(check_row_count(store)):>11}{str(check_period_aware_completeness(store)):>14}")


def without_dedupe(records):
    """The pipeline without a deduplication stage."""
    v, _ = s1_screen(records)
    v, _ = s3_consumption(v)
    summ, _ = s4_summary(v)
    return s5_serve(summ)[0], len(v)


store, _, _, _ = task_run(SOURCE, "2026-07", (120, 120, None))
print()
print("three tries , appending , pipeline without dedupe:", without_dedupe(store))

setup                     tries written   raw  consumption    north  task alert  row check  completeness
no retry , not atomic         1     120  2320         2040  -121.82        True      False         False
no retry , atomic             1       0  2200         1800     6.12        True       True          True
three tries , appending       3     440  2640         2200      0.0       False       True         False
three tries , keyed           3     200  2400         2200      0.0       False      False         False

three tries , appending , pipeline without dedupe: ({'central': 1470.73, 'north': 1320.73, 'east': 1002.64, 'west': 928.55, 'south': 1189.73}, 2440)

The third row gives the cost of the retry. The task was tried three times, the two failed tries left 120 rows each in the partition, the successful try added 200 more rows: the partition holds 440 rows instead of 200, that is, 240 rows were duplicated. Serving is still correct and the deviation of the north is zero, because the deduplication stage swallows the excess. This behavior itself was measured in the Introduction to Data Engineering course; the new information here is that the source of the duplication is not a corruption but the own mechanism of orchestration. The retry repairs the pipeline and damages the store.

The last row shows the proof of that mechanism. In the keyed write each try overwrites the same keys; after three tries the partition holds 200 rows, the duplicated rows are zero, and the raw rows stay at 2400. What makes the retry safe is not the try itself but the write being idempotent.

The last output line says how much load the deduplication carries. When the same appending store is passed through a pipeline without a deduplication stage, the consumption becomes 2440 rows instead of 2200 and serving rises in every region: the north is 1320.73 instead of 1212.55, the central is 1470.73 instead of 1321.45. That is, the zero deviation of the third row comes not from the correctness of the pipeline but from the existence of a single stage.

The task alert column shows where the alert goes silent. In the first two rows the last try fails and the task alert goes out — but that alert only says “the task failed”; it says nothing about the deviation of −121.82 in the first row, because what produces the alert is the state of the task, not the data. In the last two rows the retry succeeds, so the task alert never goes out. In the third row one sign is left: the row count check sees 2640 rows and fires. Duplication is noisy and repairable; the half-write is silent. The most dangerous setup is not the one that raises an alert but the one that stops raising it.

Summary

• The half-written partition makes serving wrong in all five regions (between −49.09 and −121.82), yet the raw rows stay at 2320: both the row count check and the period-aware completeness check are green.
• In the atomic write the partition is never written, consumption falls to 1800 instead of 2040, but the deviation stays between −3.25 and +12.80 and both checks fire. Atomicity does not make the answer correct; it makes the failure visible and bounded.
• The reason the order reverses is that serving is a period average: the cell that is not written does not enter the average, the cell that is written halfway enters it with a low number.
• The write with a three-try retry that appends leaves 440 rows in the partition instead of 200, that is, it duplicates 240 rows; in the keyed write the duplicated rows are zero. What makes the retry safe is not the try but the write being idempotent.
• That the duplicated rows do not damage serving is the work of the deduplication stage: the same store without that stage carries the consumption to 2440 rows and the north to 1320.73.
• When the retry succeeds, the task alert goes silent; only the data alerts remain, and they never see the half-write. The place where the alert goes silent is the least measured place of the pipeline.

Next Step

In three lessons the pipeline was run: its order was read from the graph, its trigger was chosen, the failed task was retried and what the failure does to serving was counted. The common gap of these three lessons is this — every measurement was made afterwards. The table that shows the half-write was built after the run was over and the result was read; the pipeline itself said nothing at that moment. What is watched while a pipeline runs, how many periods later an indicator shows a corruption and which corruption no indicator sees were never asked. The next lesson runs the freshness, volume and schema change indicators on the six corruption classes of the Introduction to Data Engineering course and counts the catch lag of each in periods.