Messaging Systems

The previous two lessons treated ingestion through pulling: the pipeline went to the source, decided for itself what to ask, and spent its own quota. 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. Queues, the publish–subscribe pattern, offsets, and delivery semantics were measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum; the procedure is not repeated here.

This lesson’s measure is two numbers: the work the partition count distributes to consumers and the order preserved per partition. They are two faces of the same setting. 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 and the pipeline loses rows.

• DI17. The code is fictional; no real messaging system, stream processor, or orchestrator is called. The source is again 200 subscribers, 12 periods, 2400 rows, seed 20260218.
• DI18. Writing to the topic goes period by period; after all subscriber records of a period have been written, the next period begins.
• DI19. A record falls into a single partition according to the partition key, and the same key always goes to the same partition. The offset within a partition preserves the write order; there is no such order across partitions.
• DI20. A partition is read by a single consumer within the group. The partition count is therefore the ceiling of the number of active consumers in the same group. The group has four consumers.
• DI21. Three partition keys are tested: subscriber, record (the combination of subscriber and period), and period. All three are fields of the source itself; none is invented.
• DI22. Each consumer computes consumption only from the records in its own partition; if the previous period of a subscriber is in another partition, that row cannot be computed and is dropped.
• DI23. The baseline is the unorchestrated run; on the healthy pipeline consumption is 2200 rows and the serving deviation is empty.
• DI24. The partition count and the idle consumers are whole numbers, no ratio is printed. The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving.

Topic, Partition, and Consumer Group

The block below sets up the common definition and adds to it a partitioned topic and a consumer group that reads it. Five numbers are printed for each setup: the row counts of the most loaded and the least loaded consumer, the number of idle consumers, the number of subscribers spread over more than one partition, and the consumption rows collected from the partitions.

# FICTIONAL. No real messaging system, stream processor, 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 topic: partitions , partition key , consumer group.
def route(key, p):
    """Partition number from the key. The same key always falls into the same partition."""
    x = 0
    for c in key:
        x = (x * 131 + ord(c)) & M32
    return int(rng(SEED + x)() * p)


def partition_records(records, p, key):
    """Writing to the topic goes period by period ; each record is written to a partition by its key.
    The offset within the partition preserves the write order."""
    partitions = {b: [] for b in range(p)}
    for k in sorted(records, key=lambda k: (k["period"], k["subscriber"])):
        value = {"subscriber": k["subscriber"], "record": k["subscriber"] + k["period"],
                 "period": k["period"]}[key]
        b = route(value, p)
        partitions[b].append(dict(k, offset=len(partitions[b])))
    return partitions


def consumer_group(partitions, c):
    """A partition is read by a single consumer within the group ; the partition count is the ceiling."""
    assignment = {t: [] for t in range(c)}
    for b in partitions:
        assignment[b % c].append(b)
    return assignment


raw = generate_source()
correct_serving, correct_n = read_result(raw)
print("consumer group 4 consumers , consumption on the healthy pipeline", correct_n, "rows")
print("key        partitions max load min load  idle split subscribers computed consumption")
for key in ("subscriber", "record", "period"):
    for p in (1, 2, 4, 8):
        parts = partition_records(raw, p, key)
        assign = consumer_group(parts, 4)
        loads = [sum(len(parts[b]) for b in assign[t]) for t in assign]
        placed = {}
        for b in parts:
            for k in parts[b]:
                placed.setdefault(k["subscriber"], set()).add(b)
        consumed = sum(len(s3_consumption(parts[b])[0]) for b in parts)
        print(f"{key:<10} {p:>10} {max(loads):>8} {min(loads):>8} "
              f"{sum(1 for y in loads if y == 0):>5} "
              f"{sum(1 for v in placed.values() if len(v) > 1):>17} {consumed:>20}")

consumer group 4 consumers , consumption on the healthy pipeline 2200 rows
key        partitions max load min load  idle split subscribers computed consumption
subscriber          1     2400        0     3                 0                 2200
subscriber          2     1200        0     2                 0                 2200
subscriber          4      708      492     0                 0                 2200
subscriber          8      696      528     0                 0                 2200
record              1     2400        0     3                 0                 2200
record              2     1211        0     2               200                 1076
record              4      621      583     0               200                  527
record              8      624      572     0               200                  268
period              1     2400        0     3                 0                 2200
period              2     1600        0     2               200                 1000
period              4      800      400     0               200                  200
period              8     1000      200     0               200                  200

The Work the Partition Count Distributes

The first pair of columns writes a fact independent of the key: in a topic with a single partition, three of the four consumers sit idle, with two partitions two sit idle, and when the count goes up to four none is left idle. The partition count is not a preference but a ceiling; adding consumers to the group gains nothing beyond the point where they exceed the partition count. At eight partitions four consumers work as well, but each reads two partitions.

The second observation is how the load is divided. With the subscriber key, four partitions skew between 708 and 492 rows, that is the most loaded consumer does 1.44 times the work of the emptiest one. With the record key the same setup is squeezed into 621 to 583: this is the most balanced of the three keys, because a key that combines subscriber and period produces 2400 distinct values and spreads almost evenly over the partitions. The period key is the worst; distributing twelve distinct values over four partitions opens the load to between 800 and 400, and at eight partitions to between 1000 and 200. Someone looking only at these three rows would choose the record key.

Order Preserved per Partition

The last two columns reverse the choice. With the subscriber key no subscriber is split and consumption stays at 2200 rows at every partition count. With the record key all 200 subscribers are split, and consumption falls to 527 rows at four partitions and to 268 at eight. The mechanism is a single sentence: a partition preserves order within itself, there is no order across partitions, and consumption is the difference of two consecutive periods of a subscriber. If the two periods fall into different partitions, there is no consumer to compute that row.

print("key        partitions computed consumption  serving deviation")
for key, p in (("subscriber", 4), ("record", 4), ("period", 2), ("period", 4)):
    parts = partition_records(raw, p, key)
    summ, consumed = {}, 0
    for b in parts:                                 # each consumer reads its own partition
        v, _ = s3_consumption(parts[b])
        consumed += len(v)
        o, _ = s4_summary(v)
        for name, s in o.items():
            summ[name] = summ.get(name, 0) + s
    s, _ = s5_serve(summ)
    print(f"{key:<10} {p:>10} {consumed:>20}  ",
          {b: round(s.get(b, 0) - correct_serving[b], 2) for b in correct_serving})

parts = partition_records(raw, 4, "subscriber")
print()
print("subscriber-keyed partition 0 :", len(parts[0]), "records , last offset", parts[0][-1]["offset"])
print("first three records :", [(k["subscriber"], k["period"], k["offset"]) for k in parts[0][:3]])
sub = parts[0][0]["subscriber"]
print(sub, "partitions of its records :",
      sorted(set(b for b in parts for k in parts[b] if k["subscriber"] == sub)))
rec_parts = partition_records(raw, 4, "record")
print(sub, "under the record key :", sorted(set(b for b in rec_parts for k in rec_parts[b] if k["subscriber"] == sub)))

key        partitions computed consumption  serving deviation
subscriber          4                 2200   {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0}
record              4                  527   {'central': -977.27, 'north': -960.55, 'east': -686.91, 'west': -616.0, 'south': -801.1}
period              2                 1000   {'central': 16.15, 'north': -7.35, 'east': 12.84, 'west': -15.29, 'south': 4.65}
period              4                  200   {'central': 180.55, 'north': 57.45, 'east': 124.64, 'west': -24.09, 'south': 25.45}

subscriber-keyed partition 0 : 492 records , last offset 491
first three records : [('A005', '2026-01', 0), ('A007', '2026-01', 1), ('A014', '2026-01', 2)]
A005 partitions of its records : [0]
A005 under the record key : [0, 1, 2, 3]

The last four lines show the mechanism on a single subscriber. With the subscriber key, all twelve records of A005 are in partition 0, and the consumer that reads that partition sees all twelve in order. With the record key, the records of the same subscriber are spread over all four partitions; no consumer sees two consecutive periods of that subscriber together. The offset within the partition keeps increasing flawlessly — partition 0 numbers its own 492 records from 0 to 491. The order has not been broken; it has been preserved in the wrong place.

The second job of the offset touches the rule of the course. A consumer group keeps a single number per partition: how far it has read in that partition. When the group fails or is set up again, it continues reading from that offset, and when the offset is moved back the same records are read again. This is the basis of making the ingestion step rerunnable; to be able to do the same with a pulling connector, the source must still be serving the old rows. Since the last offset of partition 0 is 491, reading that partition from the start brings back 492 records, and as long as the target write is idempotent, the second run gives the same result as the first.

The rule that follows is that the most balanced key may be the most wrong key. The partition key looks like a load-distribution setting, whereas it determines which records the consumer can see together. The right key is the grouping field of the computation: because consumption is computed per subscriber, the key should be the subscriber. The price of this is a load skew of 1.44 times, and it is cheap against the whole of the 2200 rows.

The Served Number Misleading

The two rows of the period key are the quietest result of the lesson. At two partitions consumption falls from 2200 to 1000 rows, that is 1,200 rows disappear, yet the serving deviation stays between −15.29 and +16.15: a movement not even reaching 2% on values of a thousand-odd units. The four-partition setup with the record key, on the other hand, does not look like it loses fewer rows — with 527 rows it loses more — and it brings the serving down by −616 to −977.

The difference lies not in the amount of the loss but in the loss’s shape. The summary layer takes a total for each region and period pair, and the serving layer takes the period mean of those totals. The period key drops a whole period at once: the cell of that period is never formed and the mean is taken over the remaining periods, each correct on its own. The record key, however, drops random rows from inside every period: the cells are formed but they total short, and the mean goes down across the board. A larger loss from the same mechanism can produce a smaller deviation. The correctness of a partitioning decision cannot be measured by looking at the served number; its measure is the consumption rows computed.

Summary

• The partition count is the ceiling of the number of active consumers in the same group: with a single partition three of the four consumers, with two partitions two of them stay idle; with four partitions all work.
• The partition key is not a load-distribution setting but a decision that determines which records the consumer can see together. A partition preserves order within itself, there is no order across partitions.
• The subscriber key splits no subscriber and keeps consumption at 2200 rows at every partition count; its price is a load skew between 708 and 492, that is 1.44 times.
• The most balanced key may be the most wrong key: the record key squeezes the load into 621 to 583 but splits all 200 subscribers and lowers consumption to 527 rows at four partitions.
• The period key moves the serving only between −15.29 and +16.15 while losing 1,200 of the 2200 rows at two partitions, because dropping a whole period does not disturb the mean. The measure of the partitioning decision is not the served number but the consumption rows computed.

Next Step

In these three lessons the data was ingested into the pipeline: the method was chosen, the constraint of the source was counted, and the topic was partitioned and distributed to a consumer group. In all three, the period was treated as a closed box — all the records of a period were counted as in hand while that period was being processed. A record can arrive after its own period has closed, and then which window will own the result becomes a decision. The next lesson measures stream processing with this question: how the window decision changes the result for a late-arriving record, and what the cost of carrying state between windows is. The common question of the remaining lessons was never asked here: whether data is lost while being ingested into the pipeline.