Competing Consumers

The “4.00 workers” in the previous lesson’s last line was not a capacity but a lower bound; at utilization of exactly one, the queue never drains. The number left a question open: does adding a worker really translate into addition? The first lesson’s backlog table treated consumer capacity as a single number and never asked where it came from — yet capacity is the product of consumer count and per-consumer rate, and one factor does not grow forever.

This lesson measures that product. Competing consumers are a set of consumers pulling from the same queue, where each message goes to only one of them. The same arrangement was built and measured in the Caching, Queues and Asynchronous Processing course, where each message also went to a single side; the pattern is the same, this course just uses the catalog name for it. The difference from the consumer groups built there is one sentence: a group decides how many separate sides the same message gets copied to; competing consumers decide how many units one side’s work gets split into. This lesson’s measure is exactly that split’s cost: the tradeoff between throughput and order.

Work That Splits, a Resource That Does Not

The setup puts 5,000 shipments’ state events into one queue. Events per shipment come from K01’s V4 assumption: 7, for a total of 35,000 events. The same shipment’s consecutive events land next to each other in the queue, because a carrier sends several scans in one transfer — the only way order can break, and the model’s focus.

There is one thing consumers cannot split: the store where an event’s effect gets written. In the model this is a parameter named the store slot — at most S events apply per tick. S is not a measured number, it is a model parameter, and its job is to show that consumer count alone is not a capacity. It runs with two values: S = 2 and S = 4.

A tick is an abstract step. Processing one event takes 1–3 ticks drawn from the seed; a shipment’s final event (“delivered”) takes 6 ticks, because it also prepares the invoice line. Every run faces the same event sequence, so the results are deterministic and machine-independent.

// competing/consumer.mjs — how many consumers pulling from the same queue affect drain time and
// the final state of shipments. MODEL: a tick is an abstract step; store slot S is a model
// parameter. Results are deterministic (computed-value class) and machine-independent.
const SHIPMENTS = 5000;            // shipments in the workload
const STEPS = 7;                   // K01 assumption V4: state events per shipment
const GAP = 1;                     // queue spacing between the same shipment's consecutive events
const SEED = 20260731;

function generator(seed) {
  let a = seed >>> 0;
  return () => {
    a = (a + 0x9e3779b9) >>> 0;
    let z = Math.imul(a ^ (a >>> 16), 0x21f0aaad) >>> 0;
    z = Math.imul(z ^ (z >>> 15), 0x735a2d97) >>> 0;
    return ((z ^ (z >>> 15)) >>> 0) / 4294967296;
  };
}

function workload() {              // event sequence in the order it lands in the queue
  const r = generator(SEED);
  const events = [];
  for (let ship = 0; ship < SHIPMENTS; ship += 1) {
    const start = r() * SHIPMENTS * GAP;
    for (let step = 0; step < STEPS; step += 1) {
      events.push({ ship, step, t: start + step * GAP + r() * GAP, duration: step === STEPS - 1 ? 6 : 1 + Math.floor(r() * 3) });
    }
  }
  return events.sort((a, b) => a.t - b.t);
}

function run(events, N, S) {
  const consumers = Array.from({ length: N }, () => null);
  const lastApplied = new Array(SHIPMENTS).fill(-1);
  const outOfOrder = new Set();
  let cursor = 0, tick = 0, applied = 0;
  while (applied < events.length) {
    for (const c of consumers) if (c && c.remaining > 0) c.remaining -= 1;
    let slotsLeft = S;              // store write slots: at most S events applied per tick
    for (let i = 0; i < N; i += 1) {
      const c = consumers[i];
      if (c && c.remaining === 0 && slotsLeft > 0) {
        slotsLeft -= 1;
        if (c.job.step < lastApplied[c.job.ship]) outOfOrder.add(c.job.ship);
        lastApplied[c.job.ship] = c.job.step;
        applied += 1;
        consumers[i] = null;
      }
    }
    for (let i = 0; i < N; i += 1) {
      if (consumers[i] === null && cursor < events.length) {
        const job = events[cursor]; cursor += 1;
        consumers[i] = { job, remaining: job.duration };
      }
    }
    tick += 1;
  }
  const wrongFinal = lastApplied.filter((k) => k !== STEPS - 1).length;
  return { tick, outOfOrder: outOfOrder.size, wrongFinal, throughput: events.length / tick };
}

const events = workload();
console.log(`workload: ${SHIPMENTS} shipments x ${STEPS} events (V4) = ${events.length} events`);
console.log(`store slots S: at most S events applied per tick (model parameter)\n`);
for (const S of [2, 4]) {
  console.log(`S = ${S}`);
  console.log("  N  drain ticks  speedup  added consumer's contribution  throughput(events/tick)  out of order  wrong final state");
  let previous = null;
  const baseline = run(events, 1, S).tick;
  for (const N of [1, 2, 3, 4, 6, 8]) {
    const r = run(events, N, S);
    const h = baseline / r.tick;
    const contribution = previous === null ? "-" : ((h - previous.h) / (N - previous.N)).toFixed(3);
    console.log(`${String(N).padStart(3)}  ${String(r.tick).padStart(11)}  ${h.toFixed(2).padStart(7)}  ` +
      `${contribution.padStart(29)}  ${r.throughput.toFixed(3).padStart(23)}  ${String(r.outOfOrder).padStart(13)}  ` +
      `${String(r.wrongFinal).padStart(18)}`);
    previous = { h, N };
  }
  console.log();
}

const PEAK = 97.22;                // K01: peak write requests/s
const PROCESSING = 48.61;          // lesson 01's chosen processing capacity (half the peak)
const soloThroughput = run(events, 1, 2).throughput;
console.log(`K01 revisited: peak write ${PEAK} events/s, lesson 01's processing capacity ${PROCESSING} events/s`);
for (const S of [2, 4]) {
  const ceiling = S / soloThroughput;
  console.log(`  S = ${S}: asymptotic ceiling ${ceiling.toFixed(2)} consumers -> a single consumer needs at least ` +
    `${(PROCESSING / ceiling).toFixed(2)} events/s (queued) / ${(PEAK / ceiling).toFixed(2)} events/s (synchronous)`);
}
const a = run(events, 6, 2), b = run(events, 8, 2);
console.log(`  S = 2, N 6 -> 8: drain ${((b.tick / a.tick - 1) * 100).toFixed(1)}%, ` +
  `out of order +${((b.outOfOrder / a.outOfOrder - 1) * 100).toFixed(0)}%, ` +
  `wrong final state ${a.wrongFinal} -> ${b.wrongFinal}`);

workload: 5000 shipments x 7 events (V4) = 35000 events
store slots S: at most S events applied per tick (model parameter)

S = 2
  N  drain ticks  speedup  added consumer's contribution  throughput(events/tick)  out of order  wrong final state
  1        89972     1.00                              -                    0.389              0                   0
  2        44987     2.00                          1.000                    0.778            102                   0
  3        30620     2.94                          0.938                    1.143            248                   0
  4        23845     3.77                          0.835                    1.468            511                   0
  6        18470     4.87                          0.549                    1.895           1590                  30
  8        17551     5.13                          0.128                    1.994           2279                 432

S = 4
  N  drain ticks  speedup  added consumer's contribution  throughput(events/tick)  out of order  wrong final state
  1        89972     1.00                              -                    0.389              0                   0
  2        44987     2.00                          1.000                    0.778            102                   0
  3        29993     3.00                          1.000                    1.167            236                   0
  4        22496     4.00                          1.000                    1.556            429                   0
  6        15099     5.96                          0.980                    2.318            915                   0
  8        11591     7.76                          0.902                    3.020           1543                   0

K01 revisited: peak write 97.22 events/s, lesson 01's processing capacity 48.61 events/s
  S = 2: asymptotic ceiling 5.14 consumers -> a single consumer needs at least 9.45 events/s (queued) / 18.91 events/s (synchronous)
  S = 4: asymptotic ceiling 10.28 consumers -> a single consumer needs at least 4.73 events/s (queued) / 9.45 events/s (synchronous)
  S = 2, N 6 -> 8: drain -5.0%, out of order +43%, wrong final state 30 -> 432

Where the Contribution Stops

The “added consumer’s contribution” column gives speedup’s per-consumer increase; a contribution of one means the added consumer did one consumer’s worth of work.

In the S = 2 column, contribution is 1.000 at N = 2, 0.938 at N = 3, 0.835 at N = 4, 0.549 at N = 6, and 0.128 at N = 8. The eighth consumer adds less than an eighth of what the first one did. Drain time goes from 18,470 to 17,551 ticks, a drop of only 5 percent. The throughput column says why: 1.994 events/tick, the S = 2 limit itself. Adding a consumer is a decision on the queue’s side, but the ceiling is set by a resource outside the queue.

In the S = 4 column, the same consumer counts behave differently: contribution stays above 0.98 through N = 6, and is 0.902 at N = 8. Speedup is 7.76, throughput 3.020 events/tick. This time the ceiling is far away, and adding a consumer still pays off. The difference between the two columns is a single parameter, and nothing changed on the queue’s side.

The ceiling itself can be calculated. Since a single consumer’s throughput is 0.389 events/tick, the asymptotic speedup ceiling is $S / 0.389$: 5.14 consumers for S = 2, 10.28 for S = 4. The measured values 5.13 and 7.76 are consistent with this ceiling. The design result is direct: the upper bound on useful consumers behind a queue is a single consumer’s rate divided by the capacity of the narrowest resource they share.

What Breaks in Exchange for Throughput

The two right-hand columns are the cost. “Out of order” means at least one of a shipment’s events is applied after a later event; “wrong final state” means a shipment’s last-applied event is not the seventh step, so it does not appear finished in the record.

With one consumer, both are zero: queue order is application order. With two consumers, 102 of 5,000 shipments lose their order. At four, 511; at six, 1,590; at eight, 2,279 — 45.6 percent of shipments. The breakage accelerates with consumer count, because as more events process at once, the odds of two of the same shipment’s events colliding rise.

The relationship between the two columns is this lesson’s most important result. At S = 4 and N = 8, 1,543 shipments lose their order, yet none end in a wrong final state. At S = 2 and N = 8, 432 of the 2,279 reordered shipments do. The difference is that the final event takes six ticks, and a narrow store slot delays applying it. So whether reordering turns into a visible error depends on the coincidence of job durations and resource scarcity. A design cannot rely on that coincidence: change the load distribution, and a number that was 0 becomes 432.

The last line reduces this tradeoff to a single line. At S = 2, going from six to eight consumers shortens drain time by 5 percent, raises reordered shipments by 43 percent, and raises wrong-final-state shipments from 30 to 432. The gain is five percent, the cost fourteen times that. Consumers beyond the ceiling add no throughput, only breakage.

The fix for preserving order — routing the same key’s events to the same partition — was built in the Caching, Queues and Asynchronous Processing course, which measured how partition count affects parallelism and within-key order there. It is not repeated here; this lesson’s contribution is placing reordering’s cost at each consumer count next to the capacity decision.

Back to the Numbers

The ceiling turns into a requirement once translated into K01’s numbers. The first lesson set processing capacity at half the peak, 48.61 events/s; the synchronous design needed 97.22 events/s.

At S = 2, the highest capacity the group can deliver is 5.14 times a single consumer’s rate. Reaching 48.61 events/s needs a single consumer processing at least 9.45 events/s; the synchronous design’s 97.22 events/s needs 18.91 events/s. If one consumer falls below that, no consumer count is enough, and the fix is not on the queue’s side. At S = 4, the same thresholds drop to 4.73 and 9.45 events/s.

The rule that follows: a capacity target is met not by “how many consumers” but by the product of two numbers, one of which depends on the shared resource’s capacity. In a design discussion, “we add consumers” said before the ceiling is calculated is not a solution — it is an assumption.

Summary

• Competing consumers decide how many units one side’s work splits into; consumer groups decide how many sides the same message gets copied to.
• An added consumer’s contribution shrinks: at S = 2, it is 1.000 at N = 2, 0.835 at N = 4, 0.128 at N = 8; going from six to eight consumers shortens drain time by only 5 percent.
• The ceiling is set by the shared resource, not the queue: the asymptotic speedup ceiling is 5.14 consumers for S = 2, 10.28 for S = 4, and the measured 5.13 and 7.76 are consistent with it.
• Reordering accelerates with consumer count: of 5,000 shipments, 102 lose their order at two consumers, 511 at four, 2,279 at eight.
• Whether reordering turns into a visible error is a matter of chance: at S = 4 and N = 8, none of the 1,543 reordered shipments produce a wrong final state, while at S = 2, 432 of 2,279 do.
• Back to K01: a processing capacity of 48.61 events/s requires at least 9.45 events/s from a single consumer at S = 2, and 4.73 at S = 4; if one consumer falls below that, consumer count is not the fix.

Next Step

This lesson’s workload was an event sequence flowing at the same rate all day, with consumer count chosen against that rate. The first lesson’s profile carried the same assumption: the peak is known, when it arrives is known, its duration is known. In a real stream none of the three is guaranteed. When the carrier’s connection drops for twenty minutes, no events arrive; then everything that piled up arrives at once, at a rate with nothing to do with the day’s peak. The next lesson takes up this spike: it measures what a queue spreading load over time means for an unpredicted peak, how large the buffer has to be, and what gets dropped when it is not enough.