--- title: 'Messaging Systems' source: 'https://academia.sh/en/courses/data-pipelines/messaging-systems' course: 'Data Pipelines and Orchestration' language: en updated: '2026-10-21T05:02:58+00:00' license: 'CC BY-SA 4.0' objectives: - 'Defines topic, partition, partition key, and consumer group through the work the partition count distributes.' - 'Shows the ceiling relation between the partition count and the consumer count by counting the idle consumers.' - 'Measures how many rows it costs the consumption computation when a partition key splits the records of the same subscriber.' - 'Explains why the partitioning with the largest row loss moves the served number the least.' --- # Messaging Systems Counting side by side the work that the partition count distributes to a consumer group and the order preserved per partition: in a group of four consumers a single partition leaves three consumers idle, four partitions put all of them to work. In a topic partitioned by the subscriber key no subscriber is split and consumption stays at 2200 rows at every partition count, but the load skews between 708 and 492. The most balanced key, the record key, squeezes the load into 583 to 621 and splits all 200 subscribers: consumption falls to 527 rows at four partitions and the serving shifts between 616 and 977 in every region. The period key, on the other hand, loses 1,200 of the 2200 rows at two partitions while moving the serving only between −15.29 and +16.15. 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. ```python # 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. ```python 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.