Scenarios

At about 200 bytes per record, with both sides buffered for 600 seconds: 1,000,000 × 200 B × 600 s × 2 = 240 GB of join state, more if both streams run at that rate.

That does not fit on a heap, so use an embedded LSM state store with incremental checkpoints, and expect checkpoint traffic to follow the changes, not the 240 GB. Use an interval join keyed by user, with an explicit bound, a watermark per partition, and handling for idle partitions.

1. Events later than the bound are lost silently: measure the real distribution of arrival delays, and count records dropped because their match was evicted.
2. A hot user key cannot be split: check key skew.
3. Compaction debt causes write stalls, which show up as backpressure.

Run at about 50–60% of capacity, so a 10-minute outage is caught up in about as long. And decide whether a 20-minute bound, which doubles the state, is worth 1–2% more matches.

Restore time = state ÷ bandwidth: at 1 GB/s, 600 seconds, 10 minutes. Incremental restore from local disk can be faster; from remote storage, slower.

The backlog is the arrival rate × 900 seconds. A job running at 90% of capacity drains at only 10% of capacity, so catching up takes 9 times the outage, 135 minutes; at 60%, it drains at 40%, taking 1.5 times the outage, 22.5 minutes. In total, restore plus catch-up.

More headroom (the recovery budget); faster restore (local recovery on the same machine where possible, and smaller state through TTLs and tighter retention rules); or replaying faster than normal by scaling out temporarily. Scaling out moves state too, so decide on it before the incident.

The input log's retention must be longer than the outage plus the catch-up.

Two-phase commit driven by checkpoints. Output is invisible until the checkpoint completes, so latency is at least the checkpoint interval. The broker's transaction timeout must exceed the longest checkpoint plus the worst recovery, or a long outage silently loses pre-committed data.

Derive a deterministic key, such as a charge ID from the order ID, and upsert, so replays overwrite the same row. No timeout coupling, output visible at once; it needs a sink that supports upserts.

Deterministic replay: no now(), no random(), no lookups whose results depend on timing.

Idempotent upserts for the charges table. Use transactions only for sinks that cannot upsert, such as append-only logs, and then set the timeout above checkpoint duration plus recovery, and document the dependency.

It is a stop and restore with state moved (downtime ≈ state moved ÷ bandwidth), followed by a catch-up that usually dominates.

It must be at least 16, and ideally divisible by both counts (for example 128 or 720), or key-group ranges get split unevenly.

An actuator that is expensive, slow and itself increases lag will oscillate. Use hysteresis (scale up at 60%, down at 30%), cooldowns, and no scaling during a backlog. Prefer scaling ahead of known peaks, during quiet hours.

State stored remotely makes rescaling cheap, at the price of a round trip per access and uncertain performance for hot loops.

The state's schema is a public interface: treat changes to it like API changes.

1. Explicit operator IDs from day one.
2. Only compatible changes, such as adding a field with a default; never remove key fields or change types without a migration.
3. A savepoint before every deployment, and a test restore in staging with a copy of real state.
4. For incompatible changes, run in parallel and cut over: a second job started from an earlier offset, double compute during the replay, source retention long enough, and idempotent sinks so the overlap is harmless.
5. Avoid starting from empty state for jobs with long windows, which would give, say, 30 days of degraded output.

Long-lived state turns every change into a migration.

1. Every bound chosen to save memory: the join interval, state TTL, allowed lateness, maximum session length. Measure the real distribution of event delays against each.
2. Watermarks: idle partitions that wake up with old events create late records, which are dropped.
3. Determinism: did something non-deterministic make a replay after a crash change the counts?
4. The sink: an append-only sink receiving a changelog, or an aborted transaction.

Widen the bound that is cutting data, add counters and alerts for records dropped as late or evicted, and reconcile against the warehouse as a routine check.

A key's state cannot be split, so throughput is capped at about the capacity of one instance ÷ 0.3 ≈ 3.3 instances' worth, however many you add.

Counting is associative and commutative, so aggregate in two phases: salt the key into, say, 8 sub-keys, count locally, strip the salt, then sum. It costs an extra shuffle and a short additional stage of latency.

For order-sensitive or non-decomposable operations, such as a pattern match or an exact median: reduce the work for that key at the source (pre-aggregate in the producer), or accept the cap.

Per-key skew, so the problem shows up before it hits the limit.

That is 86,400 overlapping windows, impossible naively. Panes of 1 second help, but combining 86,400 panes per output is costly, so use larger panes (say 1 minute) and a coarser refresh, or a structure that merges cheaply.

Not decomposable exactly; use HyperLogLog sketches, which merge.

A transactional sink adds latency of at least the checkpoint interval, which conflicts with refreshing every second, unless the sink is an idempotent upsert on (restaurant, refresh time).

Approximate distinct counts, 1-minute panes, idempotent upserts, and a clear statement of which numbers are approximate.