A single 35% merchant defeats parallelism increases
merchant_metrics computes 5-minute tumbling SUM(amount) and COUNT(*) grouped by merchant_id from a 36-partition Confluent Cloud topic and writes the results to Kafka. Job parallelism is 24, checkpointing is every 5 minutes, and the source has a stable watermark. One merchant, m_1042, is a marketplace aggregator and accounts for about 35% of all events. Subtask 2 is pinned near 100% busy while the other 23 sit at 25-40%; checkpoint alignment takes 60-90 s and checkpoints frequently run past the interval. Consumer lag on the source grows by roughly 3,000 records/s. There is no GC pressure, CPU is not saturated overall, and RocksDB reads and writes are not the bottleneck. The team doubles parallelism to 48: end-to-end throughput is unchanged, the same lopsided pattern reappears, and checkpoint size grows by about 1.7x. Watermarks advance normally and no records are being dropped as late.
1. Why did doubling parallelism fail to add throughput? Identify precisely where the bottleneck sits and explain the 1.7x checkpoint growth.
Read the worked answer
After GROUP BY merchant_id, Flink hashes the key to choose a subtask. m_1042 is a single key, so all of its 35% lands on exactly one subtask no matter how many subtasks exist; the remaining 65% spreads across the others. End-to-end throughput is set by the slowest instance, so adding parallelism allocates more resources without removing any load from the hot instance. The 60-90 s checkpoint alignment is the symptom, not the cause: barriers wait on the one channel that is continuously busy. Checkpoint growth is expected - total keyed state is unchanged, but there are twice as many RocksDB instances, more SST files, more open handles and more per-subtask overhead per snapshot. Note also that local-global aggregation and mini-batch do not help here: the aggregation functions are cheap, so the cost is the hash shuffle and the per-key state, not the computation. Diagnosing this from per-subtask numRecordsInPerSecond is the decisive step.
2. Rewrite the aggregation to remove the hot spot. How do you choose the salt count, and what does the rewrite cost?
Read the worked answer
Use two-phase aggregation. The inner stage groups by merchant_id, salt, where salt is something like MOD(ABS(HASH(order_id)), N) or a precomputed bucket column; the outer stage groups by merchant_id alone and sums the partial sums and counts. That spreads one hot key over N subtasks. Size N from the hot-key share: with 35% on one key and 24 subtasks, the average load per subtask is 100/24 ≈ 4.2%, so N=8 leaves roughly 35/8 ≈ 4.4% per salted subtask, which is close to the mean. Costs: an extra shuffle and an extra network exchange, plus N partial rows of state per hot key instead of one, and more complex SQL that must stay in sync with any other per-merchant logic. Additive aggregates such as SUM and COUNT compose cleanly; COUNT(DISTINCT ...) does not and needs table.optimizer.distinct-agg.split.enabled or a separate pipeline. Keep the salt deterministic so reprocessing from a checkpoint produces identical results.
3. What else would you consider besides salting, and what evidence would convince you the fix worked?
Read the worked answer
Alternatives worth weighing: route m_1042 into a dedicated job with its own parallelism and union the results downstream - a simpler SQL change but two deployments and two failure domains to operate; pre-aggregate to (merchant_id, minute) granularity before the rollup to cut shuffle volume if the dashboard tolerates minute-level intermediate results; or reconsider the key entirely if the metric is really per-region rather than per-merchant. Increasing sink partitions is irrelevant because the bottleneck is upstream of the sink, and rebalance() does not help either - the exchange already round-robins, and the hash on the merchant key is what funnels traffic onto one instance. Evidence of success: per-subtask numRecordsInPerSecond and numRecordsOutPerSecond become roughly flat, checkpoint alignment drops below a few seconds, checkpoints complete well inside the 5-minute interval, and source lag stops growing and then drains. If alignment stays high while subtask load is flat, look at the sink next.