Connectors and API Consumption

The previous lesson treated the ingest task as a single move: the current state of the source was called and the rows were in hand. Source systems are not read like this. A source that serves data through an endpoint delivers it page by page, limits how many calls can be made in a given window, and some of the calls drop temporarily. A single run of the ingest task is no longer one operation but a sequence of dozens of calls.

This lesson’s measure is two numbers: the calls completed under the constraint and the rows received. Pagination and rate limiting themselves were measured in the Web API Design, SQL Fundamentals, and Building Applications with Model APIs courses; the procedure is not repeated here. The only question asked here is this: under these constraints, how many of the rows the pipeline wants can a connector bring in, and what trace does the row it fails to bring leave behind.

• DI9. The code is fictional; no real endpoint, connector product, or orchestrator is called. The source is again 200 subscribers, 12 periods, 2400 rows, seed 20260218.
• DI10. The endpoint serves the rows of a period in pieces of page size and reports how many rows the period holds; the connector therefore knows in advance how many pages to request.
• DI11. The rate limit is modeled as a quota per window: each period has one window, and at most quota calls can complete in that window. A page that does not fit is left to the next window; that is a delay of one period. Time is not written as raw duration.
• DI12. The transient failure rate is 0.15 and is derived from the seed; the same attempt at the same page gives the same result on every run. The pipeline is deterministic.
• DI13. The retry allowance is eaten from the quota: the repeat of a failed call is also a call. The backoff procedure was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and is not repeated here.
• DI14. A page whose allowance runs out is abandoned and the connector moves on; the rows of that page are never received.
• DI15. The baseline is the unorchestrated run: no retry, no queue, a single manual trigger. Retry is the only mechanism this lesson adds.
• DI16. The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving.

A Connector Under Constraint

The block below sets up the common definition and adds to it a paginated endpoint and a connector that consumes it. The connector has a single queue: in each period the pages of that period enter the queue, pages are pulled until the window quota runs out, and the rest roll over to the next window.

# FICTIONAL. No real endpoint, connector product, or orchestrator is called;
# the meter readings of the measurement network are fictional, seed 20260218.
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():                              # one reading per subscriber per period
    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


# ---- PIPELINE: five stages. Each stage counts the rows it drops.
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):                              # per-region mean over the periods
    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)


# ---- FICTIONAL endpoint: paginated read , quota per window , transient failure.
def page(records, period, number, page_size):
    rows = [k for k in records if k["period"] == period]
    return rows[number * page_size:(number + 1) * page_size]


def call_fails(period, number, attempt, failure_rate):
    """The transient failure is derived from the seed , it is independent of the run."""
    r = rng(SEED + 13 * PERIODS.index(period) + 3 * number + 101 * attempt)
    return r() < failure_rate


def connector(records, page_size, quota, failure_rate=0.15, retries=0):
    """Every period has one window ; at most `quota` calls complete in a window.
    A page that does not fit is left to the next window , that is a delay of one period."""
    count = (200 + page_size - 1) // page_size
    received, completed, failed, delayed, queue = [], 0, 0, 0, []
    for p in PERIODS:
        queue += [(p, n) for n in range(count)]
        remaining = quota
        while queue and remaining > 0:
            qp, n = queue[0]
            passed, attempt = False, 0
            while attempt <= retries and remaining > 0:
                remaining -= 1
                if call_fails(qp, n, attempt, failure_rate):
                    failed += 1
                    attempt += 1
                    continue
                completed += 1
                passed = True
                break
            if passed or attempt > retries:         # either received or the allowance ran out
                queue.pop(0)
                if passed:
                    received.extend(page(records, qp, n, page_size))
        delayed += len(queue)                       # pages waiting one more window
    return received, completed, failed, delayed, len(queue)


raw = generate_source()
correct_serving, correct_n = read_result(raw)
print("quota 8 calls/window , transient failure rate 0.15")
print("size  pages retries  expected completed failed delayed unfetched received consumption")
for page_size in (25, 50, 100):
    for retries in (0, 2):
        rec, completed, failed, delayed, left = connector(raw, page_size, 8, 0.15, retries)
        s, t = read_result(rec)
        count = (200 + page_size - 1) // page_size
        print(f"{page_size:>4} {count:>6} {retries:>7} {12 * count:>9} {completed:>9} {failed:>6} "
              f"{delayed:>7} {left:>9} {len(rec):>8} {t:>11}")

quota 8 calls/window , transient failure rate 0.15
size  pages retries  expected completed failed delayed unfetched received consumption
  25      8       0        96        83     13       0         0     2075        1575
  25      8       2        96        81     15     112        15     2025        1825
  50      4       0        48        39      9       0         0     1950        1300
  50      4       2        48        48     12       0         0     2400        2200
 100      2       0        24        20      4       0         0     2000        1400
 100      2       2        24        24      6       0         0     2400        2200

Page Size and the Rows Missed per Call

The three rows without retry give the first trade-off. With a page size of 25 a period is eight pages, and twelve periods ask for 96 calls; raising the size to 100 brings the same job down to 24 calls. The call count falls in inverse proportion to the page size, and against a source with a rate limit this is a direct gain.

The price of the gain is written in the same rows. At size 25, 13 calls drop and 325 rows are missed; at size 50, 9 calls drop but 450 rows are missed; at size 100, only 4 calls drop and again 400 rows are missed. The product holds exactly in every row: the rows missed are the number of dropped calls times the page size. Quadrupling the page size cuts the call count to a quarter and quadruples the cost of a single failure. The connector setting is chosen at the point where these two curves cross, and which one weighs more depends on the failure rate of the source.

The consumption column shows how the cost grows inside the pipeline. At size 50, 450 rows are missed but the consumption rows fall from 2200 to 1300: each missed row not only lowers the consumption of its own period, it also makes the consumption of the next period uncomputable, because consumption is a difference computation. A shortfall of 450 rows turns into a loss of 900 rows.

Retry Eating the Quota

The most rows a window can carry is the product of the quota and the page size. With a quota of eight, size 25 gives a ceiling of 200 rows per window, size 50 gives 400 rows, and size 100 gives 800 rows. The demand of a period, though, is exactly 200 rows. At size 25 the ceiling and the demand are exactly equal and the connector has no slack at all; at size 50 the ceiling is twice the demand, and at size 100 it is four times. What decides whether retry can find room for itself is this slack, not the failure rate.

In the size 50 and size 100 rows retry is nearly free: because 48 and 24 calls stay under the quota, failed calls can be repeated and the rows received rise to 2400 and the consumption to 2200. The deviation is zeroed.

In the size 25 row the same mechanism turns around. The expected 96 calls are already at the exact limit of the quota; because retries eat from the quota, a few pages cannot keep up in each window and the queue grows. The result is 2075 rows received without retry against 2025 rows received with retry. Adding the mechanism has lowered the rows received. Two more numbers sit beside it: 112 pages have been delayed by at least one window, and 15 pages are still waiting in the queue when the twelve windows are over. This is precisely the second claim of this course: the orchestration mechanism also breaks things, and only the one who looks at its own counter sees what it broke.

The Connector’s Own Call Ledger

The next block puts three setups side by side and asks where the loss falls.

FULL_SET = set((b, d) for b in REGION for d in PERIODS)
print("size quota retries received row ratio partition_period  received rows per period")
for page_size, quota, retries in ((25, 8, 0), (25, 8, 2), (25, 12, 2)):
    rec, completed, failed, delayed, left = connector(raw, page_size, quota, 0.15, retries)
    counts = [sum(1 for k in rec if k["period"] == p) for p in PERIODS]
    seen = set((k["region"], k["period"]) for k in rec)
    print(f"{page_size:>4} {quota:>5} {retries:>7} {len(rec):>8} {(2400 - len(rec)) / 2400:>9.4f} "
          f"{str(seen != FULL_SET):>16}  {counts}")
print()
for page_size, quota, retries in ((25, 8, 0), (25, 8, 2), (25, 12, 2)):
    s, t = read_result(connector(raw, page_size, quota, 0.15, retries)[0])
    print(f"size {page_size} quota {quota} retries {retries}",
          {b: round(s.get(b, 0) - correct_serving[b], 2) for b in correct_serving})

size quota retries received row ratio partition_period  received rows per period
  25     8       0     2075    0.1354            False  [175, 175, 175, 150, 175, 125, 200, 175, 200, 175, 150, 200]
  25     8       2     2025    0.1562             True  [200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 25, 0]
  25    12       2     2400    0.0000            False  [200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200]

size 25 quota 8 retries 0 {'central': -373.72, 'north': -302.19, 'east': -278.0, 'west': -190.91, 'south': -367.28}
size 25 quota 8 retries 2 {'central': -124.85, 'north': -118.25, 'east': -82.06, 'west': -78.39, 'south': -105.05}
size 25 quota 12 retries 2 {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0}

The period sequence in the second row is the most counterintuitive result of the lesson. The setup with retry receives fewer rows but its serving deviation is smaller: −125 … −78 instead of −374 … −191. The reason lies not in the amount of the loss but in its place. In the setup without retry the loss is scattered over twelve periods and pulls the total of every period down; in the setup with retry the first ten periods have been received completely, 2026-11 has received only 25 rows because the queue grew, and 2026-12 has not been received at all. Because the region mean is computed over the periods that exist, the serving looks repaired. An improvement in the number served does not mean the data has improved.

The answer to who reports the loss also splits across the two rows. For the scattered loss, the period-aware completeness check stays green, because every region is still visible in every period; the only thing that catches it is the row count check, at a ratio of 0.1354. For the queue loss the same check fires, because there is a period with no rows at all. The cheaper of the two is not inside the pipeline but in the connector’s own ledger: against the expected 96 calls, 83 have completed, and this difference is known without waiting for any stage, the moment the calls finish. Raising the quota from eight to twelve produces the last row: all 96 calls complete, 2400 rows are received, the deviation empties, and no delayed page remains.

Summary

• A connector’s work is counted under constraint: a page size of 25 needs 96 calls, a page size of 100 needs 24; the call count is inversely proportional to the page size, and against a source with a rate limit this is a direct gain.
• The missed rows are the number of dropped calls times the page size: 13 calls 325 rows, 9 calls 450 rows, 4 calls 400 rows. Enlarging the page size raises the cost of a single failure by the same ratio.
• The missed rows grow inside the pipeline: because consumption is a difference computation, a shortfall of 450 rows turns into a loss of 900 rows.
• Under the quota retry is free and brings back all 2400 rows; at the limit of the quota it eats the quota, delays 112 pages, and lowers the rows received from 2075 to 2025.
• When the loss collects at the tail of the queue the serving deviation shrinks (−125 … −78) but the last two periods collapse: 2026-11 receives only 25 rows and 2026-12 is not received at all. The row count does not see this, the period-aware completeness check does, and the cheapest report comes from the connector’s call ledger without delay.

Next Step

The connector in this lesson pulled the source: it decided for itself when to read, spent its own quota, and reported what it missed from its own ledger. A mechanism in which the source pushes the data on its own turns this balance around. Records are written to a topic, the topic is split into partitions, and a consumer group reads the partitions. The next lesson measures this mechanism through two numbers: the work that the partition count distributes among consumers, and the order preserved per partition. The trade-off to be measured is this: as the partition count grows, parallelism grows, but when the records of the same subscriber fall into different partitions, the order on which the consumption computation relies breaks.