Event Ingestion and Stream Processing

“Results must be fresh within five minutes” is not yet an objective. Five minutes from what event to what observation? For which records and keys? Does an offline producer count while disconnected? Is a provisional window result acceptable? When a correction arrives late, does freshness restart? Does a sink commit count before a user can query it?

A useful objective names two instants and a population:

For 99.9% of valid order events durably accepted by the ingestion gateway, the corresponding market aggregate is query-visible within 300 seconds of the event timestamp, measured over ten-minute windows. Events accepted over two minutes after their event timestamp enter the late data policy and are reported separately. Withdrawals have a stricter correction objective.

Even that statement needs an error budget, correctness rule, duplicate definition, and behavior during declared producer disconnection. But it can shape a system. The five minutes can be allocated as an operating ledger:

producer and gateway transit       25 s
durable ingestion                  10 s
queue-age allowance               120 s
event-time hold and processing     90 s
sink commit                        35 s
serving visibility                 20 s
                                    -----
allocated objective               300 s

This is a budget over one event’s path, not a license to add independently measured p99 values. Trace or correlate the same event population from acceptance through visibility; report each component distribution and the end-to-end distribution. The budget reveals which control must activate before the user-visible objective is spent.

The central design rule follows: treat lag age, recoverability, and state movement as primary capacity dimensions; record rate alone is insufficient.

Give every stage one job and one boundary

SignalWeave, the running system, turns commerce events into fraud features, customer notifications, market aggregates, and an analytical landing stream. Its continuous path has six roles.

Producers create events with a stable event identity, schema identity, event timestamp, partition key, and causal/business version. They buffer only within a bounded local policy; “the SDK will retry forever” is not durability.

Ingestion gateways authenticate, authorize, validate size/schema, apply tenant and producer quotas, assign an ingest timestamp, and acknowledge at a named boundary. They should not perform unbounded business enrichment before durable acceptance.

Durable logs append records to ordered partitions, replicate them according to the acknowledgment contract, retain enough history for consumer recovery and supported replay, and expose positions. A log is an authority for accepted event history only if its durability and retention contract says so.

Processors read partitions, transform records, repartition when keys change, maintain operator state, advance time, checkpoint progress, and produce outputs. Their input position alone does not describe the state or side effects associated with that position.

State stores hold keyed aggregates, joins, timers, deduplication identities, and window contents. State can be local with durable checkpoints/changelogs, remote, or hybrid. Name its authority and rebuild path.

Sinks commit results to serving databases, logs, indexes, object storage, or external effects. The sink determines whether replay is harmless, duplicative, or irreversible.

An end-to-end trace should carry event ID, producer and ingest times, source partition/position, processor job/version, checkpoint epoch, output identity, sink commit, and visible version. Avoid putting raw high-cardinality identities into every metric; exemplars and sampled trace indexes preserve drill-down without exploding the telemetry system.

Schema, key, and ordering are one design decision

An event is an immutable claim that something happened, not a bag of fields. Define:

event type and semantic version
stable event identity and producer identity
entity/business key and partition key
event time, ingest time, and source sequence/version
required fields, defaults, units, and null meaning
privacy/security class and retention
correction, retraction, and deletion semantics

Ordering is scoped. A partitioned log can order records within one partition while making no total-order promise across partitions. If all events for one order use order_id, an order-local processor can reason about sequence. If a later stage keys by customer_id, records cross a shuffle and the customer-keyed operator sees a different partitioned order. A global order would create a global serialization boundary and is rarely the real requirement.

Choose the key from state locality and correctness:

• order_id localizes order state and per-order sequence;
• customer_id localizes customer features but may create whale-customer skew;
• market_id localizes market aggregates but can collapse a country into one hot partition;
• random distribution balances rate but destroys key-local state and ordering.

A composite or hierarchical key can distribute a large domain, but then downstream aggregation must merge shards. That is valid only when the operation is decomposable and correction semantics remain explicit.

Schema compatibility is more than whether a decoder can parse bytes. A new field with a default may be wire-compatible while changing business meaning. Renaming an enum, changing units, altering event-time source, or reusing a field for a new identity can corrupt historical replay. Maintain reader/writer schema tests, semantic fixtures, version populations, oldest retained schema, and the job/sink versions that can interpret each one.

Burst tolerance is storage plus a control response

SignalWeave receives 50,000 records per second in steady state. Compressed records average 1.2 KiB. A scheduled promotion can produce 85,000 records per second for one hour. Base processors sustain a measured 65,000 records per second for the defined operator graph and state population.

If processing stays fixed during the burst:

backlog growth = 85,000 - 65,000
               = 20,000 records/s

one-hour backlog = 20,000 × 3,600
                 = 72,000,000 records

steady drain margin = 65,000 - 50,000
                    = 15,000 records/s

post-burst drain time = 72,000,000 / 15,000
                      = 4,800 s = 80 min

The log may tolerate that burst while the five-minute freshness objective does not. Burst absorption and freshness are different promises.

SignalWeave keeps the base 65,000-record/s processing envelope but has prevalidated recovery capacity of 105,000 records per second. Its control detects sustained arrival over base capacity and supplies that capacity after a modeled 120-second delay. During the delay, backlog reaches 2.4 million records. A simple FIFO approximation gives queue age of about 2,400,000 / 65,000 = 36.9 seconds at the scaling boundary. Once 105,000 capacity is active during the 85,000-record/s burst, the 20,000-record/s recovery margin drains the backlog in 120 seconds. Catch-up completes four minutes after burst onset under these constant assumptions.

This model supports a design, not a guarantee. Startup can involve image pulls, partition assignment, state restore, cache warm-up, connection limits, checkpoint alignment, and sink admission. Recovery capacity may process old records more slowly because state/cache locality differs. Validate the entire control loop under a production-shaped partition and state distribution.

Producer backpressure must activate before gateway memory becomes the queue itself. When durable-ingest or tenant capacity is exhausted, SignalWeave returns a classified overload outcome with retry-after guidance and a maximum acceptable event age. Cooperative producers reduce concurrency and rate, retain events only inside a bounded local byte/age budget, and expose the oldest unsent event. A producer that cannot slow—such as a physical sensor or packet tap—needs an explicit choice among durable edge spill, a separately provisioned emergency path, sampling, or loss with counters. Infinite SDK retries merely move an unbounded queue to every caller and can replay a synchronized surge after recovery. The gateway reserves control and correctness-critical capacity, rejects before its own queue consumes the freshness budget, and prevents one tenant’s buffer from becoming everyone else’s delay.

One-hour tolerance also requires retained bytes. At 85,000 records/s and 1.2 KiB:

logical one-hour burst = 85,000 × 3,600 × 1.2 KiB
                       ≈ 350.19 GiB

three log copies       ≈ 1,050.57 GiB
                       ≈ 1.026 TiB

Add segment/index overhead, compression variance, replica catch-up, retention overlap, safety margin, and other topics. Storage capacity does not substitute for processing or sink capacity, but without it the recovery plan has nothing to replay.

Lag has records, bytes, and age

An offset difference answers how many positions separate a consumer from a log head. It does not directly answer how old the result is. One partition may contain tiny rapid records; another may contain large sparse records. A producer can pause while an old event arrives. A processor can be at the head while a sink queue holds results.

Track at least:

• position lag and byte lag by partition;
• oldest unprocessed ingest age;
• oldest unprocessed event-time age;
• watermark delay by input and operator;
• processor queue/service and busy/backpressured time;
• checkpoint duration, age, size, and failures;
• sink commit and query-visible age; and
• recovery time and drain margin at current arrival.

Queue age is often the best overload control because it connects backlog to the freshness budget. Position lag is still operationally useful for retention and replay. Byte lag predicts transfer work. Event-time age reveals old valid data. No one metric replaces the others.

For variable rates, use a backlog balance rather than a single utilization number:

B(t + Δt) = max(0, B(t) + arrivals(t, Δt) - completions(t, Δt))

drain time lower bound = backlog / (sustainable capacity - arrival rate)

The denominator must be positive. Sustainable capacity is measured for the active operator graph, state, schema mix, partition skew, sink, and failure state—not a framework benchmark.

Stateless and stateful operators have different recovery costs

A stateless map or filter can restart from an input position if its output boundary is replay-safe. Its scale-out cost is primarily code, connections, partitions, and warmed dependencies.

A stateful aggregate, join, session, pattern, or deduplicator must also restore or rebuild state consistent with input progress. State includes values, timers, window membership, join buffers, deduplication identities, and sometimes pending sink transactions. Scaling can move that state to new owners.

For each operator, record:

State size is not just heap. SignalWeave has 80 million active keys at a modeled 180 bytes of logical operator state per key, about 13.41 GiB. Two durable checkpoint copies represent 26.82 GiB before serialization overhead, indexes, incremental-history references, local working copies, transfer buffers, and backend amplification. Measure full and incremental checkpoint bytes, restored bytes, state access service demand, and state movement during rescale.

State TTL needs the same caution as cache TTL. Expiring a deduplication identity before the latest possible replay permits a duplicate. Expiring join state before allowed lateness loses a match. Retaining every key forever makes checkpoint and recovery grow without bound. Tie retention to source replay, late/correction, effect, privacy, and recovery horizons.

Event time, windows, and watermarks make incompleteness explicit

Event time says when the source-domain event occurred. Processing time says when an operator runs. Ingest time marks a system boundary. They answer different questions.

A window groups events by time and key: fixed/tumbling, sliding, session, or custom. Its result is not inherently final when processing time reaches the window end because events can be delayed or out of order. A watermark is a progress estimate or assertion about event time, used to decide when to emit, update, or retire state.

Watermarks are not wall clocks and not proof that no older event will arrive unless the source contract makes that proof possible. With multiple inputs, the slow or idle partition can hold back progress. Ignoring an idle input can restore latency but risks classifying its later records as late; that choice needs an idleness and reactivation rule.

SignalWeave chooses three output states for market aggregates:

1. early/provisional updates every 20 seconds for dashboard responsiveness;
2. on-time output when the watermark passes the window end; and
3. corrected upsert for accepted late events within two hours.

The sink key includes aggregate identity and window, while the value carries output version, completeness state, event-time interval, watermark, and generated-at time. A correction replaces the prior version rather than adding a second total. Events later than two hours enter a review/reconciliation stream; withdrawals use a separate stricter path.

The design must say what happens to very late records: drop with evidence, side-output, reopen and correct, compensate, or recompute offline. “Allowed lateness = two hours” is incomplete without state retention, sink update, user-visible labeling, and backfill behavior.

Checkpoints join input position to operator state

A useful checkpoint captures a consistent relation between source positions and state so recovery can resume without inventing a state/input combination that never existed. Its practical cost includes barrier or coordination behavior, state serialization, incremental bookkeeping, durable writes, metadata, retention, and restore.

Checkpoint frequency trades replay work against steady overhead. A 30-second interval does not imply 30-second recovery: a large state may take minutes to materialize, download, deserialize, assign, warm, and replay. A completed checkpoint does not replace source retention; the source still needs records after the captured positions, and often older records for rollback or backfill.

Track:

• checkpoint trigger-to-complete distribution and oldest successful age;
• full versus transferred bytes and referenced base files;
• alignment/backpressure contribution where applicable;
• failures and consecutive failure age;
• durable-store throughput and throttling;
• restore phase time, reassignment, and replay rate; and
• compatibility across job/state schema versions.

Test recovery by killing workers, coordinators, state storage access, and sinks at defined positions. Verify not only that the job runs again, but that output identities, aggregates, and effects match a failure-free oracle for the scoped contract.

Replay does not make side effects exactly once

Processing can be replayed. External effects may not be reversible. A processor that sends an email and then crashes before recording progress can send it again. One that records progress first and crashes before sending can lose it.

Choose an effect boundary:

Idempotent/upsert sink. Derive a stable output key and version so replay produces the same terminal representation. This works for tables and materialized views that support conditional or transactional upsert semantics.

Transactional log-to-log boundary. Atomically publish output records and input progress within one supported transaction domain. Downstream external effects remain a separate boundary.

Durable effect ledger/outbox. Record intended effect identity and state in an authority, dispatch it, and reconcile ambiguous outcomes. The external provider needs an idempotency key or lookup for the strongest result.

At-least-once with explicit duplicates. Sometimes duplicate-safe consumers and cost bounds are acceptable. Say so and measure duplicates.

Deduplication identity must match the logical effect, not merely a transport delivery. Retention must cover producer retries, log retention, checkpoint rollback, backfill, disaster recovery, and external reconciliation. A Bloom filter or cache can accelerate the lookup but cannot be the only authority if false positives or loss would violate the effect contract.

Avoid the unbounded phrase “exactly once.” Name the source positions, processor state, output log/table, and external effects included; then name the failures and retention horizon excluded.

Repartitioning spends network, disk, state, and time

If input is keyed by order_id and an aggregate needs customer_id, records must be redistributed. A shuffle serializes records, sends them across a network boundary, queues them, and writes/updates newly keyed state. It can add skew, backpressure, retry, spill, and failure coupling.

At steady 50,000 records/s and 1.2 KiB, SignalWeave ingests about 58.59 MiB/s. Three log copies account for about 175.78 MiB/s of storage-network write before protocol and index overhead. If 40% of records cross a repartition boundary, payload shuffle is another 23.44 MiB/s. A 0.35 KiB sink representation adds about 17.09 MiB/s. Those flows may traverse different links, but each must be placed on a capacity map.

Record amplification by stage:

input records and compressed bytes
producer batches and retries
log replica bytes and retention writes
consumer read/decompression bytes
shuffle records/bytes and skew
state reads/writes and checkpoint bytes
output records and sink write amplification
replay/backfill multiplier

Adding source partitions raises partition parallelism only if keys distribute, consumers and state can move, and sinks accept the extra concurrency. Too few partitions cap parallelism; too many increase metadata, connections, files, scheduling, checkpoints, and small-batch overhead. A hot key remains one keyed-state owner unless the aggregation can be decomposed into partials and merged.

Repartitioning an existing stateful job is a migration. Version the key function, preserve or rebuild old state, route old and new schemas deliberately, validate dual results, and keep rollback positions. A silent hash or serialization change can move every key and make old checkpoints unreadable.

Backfills are competing production traffic

A backfill replays historical input through new or corrected logic. It can consume source read bandwidth, decompression CPU, shuffle, state I/O, checkpoints, sinks, and downstream quotas far faster than live traffic. It can also advance event time differently and produce corrections against already-visible results.

Treat reprocessing as a separate workload identity, not a flag on the live consumer. Separate identities, quotas, state/checkpoint namespaces, output versions, and dashboards provide workload isolation and make pause, rollback, and cost attribution possible. Isolation does not require permanently idle hardware; it requires live traffic to retain enforceable priority and reclaim borrowed resources quickly.

SignalWeave must process 2.4 billion historical records without starving live traffic. The validated processing envelope is 105,000 records/s. It reserves 90,000 records/s for live demand, leaving a guaranteed maximum of 15,000 records/s for backfill during the 85,000-record/s peak and 5,000 records/s of live reserve. At 15,000 records/s, the lower-bound backfill duration is:

2,400,000,000 / 15,000 / 3,600 ≈ 44.44 h

When live traffic is 50,000 records/s, backfill may borrow up to 55,000 records/s and the arithmetic lower bound falls to about 12.12 hours. Borrowed capacity is preemptible. The controller reduces backfill when live queue age, watermark delay, checkpoint age/duration, sink latency, or source/replica load crosses a guard.

The plan uses:

• a separate consumer/job identity and quotas;
• explicit source range, schema/job version, and output namespace;
• checkpointed progress by partition/range;
• deterministic output identity and correction precedence;
• token budgets for read, processing, shuffle, state, and sink—not only records;
• live-reserved capacity with fast reclaim;
• pause and abort thresholds;
• sample and aggregate comparison before promotion; and
• reconciliation after completion.

Do not merge backfill and live progress blindly. Backfill event times can make watermarks jump or stall, old schemas can trigger different code, and historical output can overwrite newer truth unless version/precedence rules prevent it.

Evolve schemas across retained history

Continuous systems run several time populations at once: new producers, lagging producers, retained old log segments, checkpointed state written by an older job, and sinks expecting one or more versions. A deployment that reads only today’s schema is not replayable.

Use compatibility gates at four boundaries:

1. producer writer schema to ingestion validation;
2. retained writer schemas to processor reader schema;
3. old checkpoint/state serializer to new job version; and
4. output schema/version to sink and serving readers.

Prefer additive fields with stable defaults when semantics permit, but test meaning with historical fixtures. For breaking changes, use a new event type or versioned topic/stream, dual-read or translate under a controlled horizon, compare outputs, and retire only after source retention, backfill, checkpoints, and consumers no longer need the old form.

Deletion and privacy requests need lineage. Removing an event from a serving table may leave logs, state, checkpoints, object storage, derived aggregates, and backfill inputs. Define lawful retention, cryptographic or physical erasure where required, recomputation, and audit separately from normal schema evolution.

Stream-processing SLO sheet

Boundary and population:
  accepted event types, valid/invalid rule, tenants/keys, exclusions

Freshness:
  event/ingest start, durable acceptance, visible end, percentile/window,
  provisional/on-time/corrected states, late-data policy

Correctness and effects:
  ordering scope, duplicate identity, output identity/version,
  effect boundary, correction/retraction, reconciliation

Traffic and partitions:
  records/s and bytes/s, burst shape/duration, schemas, key/skew,
  partition count, hottest key/partition, growth

Topology and durability:
  gateway acknowledgment, log replicas/retention, processor graph,
  state authority, sink commit and query visibility

Capacity:
  ingest/log/read/shuffle/state/checkpoint/sink demand,
  base/recovery envelope, scale delay, headroom and cost

Lag and recovery:
  position/byte/age limits, drain margin, oldest checkpoint,
  restore/replay phases, source-retention margin, recovery objective

Time and windows:
  event-time source, watermark/idleness rule, window and trigger,
  allowed lateness, state retention and correction

Backfill and evolution:
  range/version/output namespace, reserved/borrowed quotas,
  pause/abort, compatibility populations, validation/promotion

Evidence and ownership:
  event-linked traces, partition metrics, raw fixtures, drills,
  controller and on-call owner, transfer limits

Reject an SLO sheet that reports consumer offsets without visible age, claims burst tolerance without durable bytes and drain time, names checkpoints without restore tests, or calls a side effect exactly once without an identity and authority boundary.

Validation campaign

Validate the model through normal, skewed, overloaded, failed, and recovering states:

• representative records, sizes, keys, schemas, and producer timing;
• one-hour burst plus control-loop delay and loss of recovery capacity;
• hot partition/key and idle input effects on watermarks;
• processor crash before/after checkpoint and sink commit;
• state-store slowdown and checkpoint storage throttling;
• sink rejection, ambiguity, and deduplication retention boundary;
• source partition expansion and stateful rescale;
• old-schema replay from the oldest supported checkpoint/log position;
• backfill borrow/reclaim under a live burst; and
• regional recovery with restored positions, state, outputs, and effects.

Compare event-linked end-to-end freshness with stage budgets. Verify aggregate and effect correctness against an independent oracle. Record environment, code/schema versions, partition map, state size, checkpoint age, cache state, warm-up, raw data, and uncertainty. A synthetic framework throughput number does not transfer until its record, key, operator, state, sink, and failure shapes match the claim.

The fixture at examples/performance-engineering-system-design-handbook/part-05/event-streaming/ reproduces the freshness allocation, burst backlog, recovery timing, byte flows, state size, and backfill bounds.

Applied work

Five-minute, one-hour design. Starting from SignalWeave’s 50,000 steady and 85,000 burst records/s, draw the gateway, log, processor, state, and sink boundaries. Allocate the 300 seconds, specify acknowledgment and visibility, choose keys/partitions, size one-hour retained bytes, and prove that scale delay plus recovery keeps queue age within budget. Then remove recovery capacity for 20 minutes and state which objective fails first and how the system degrades.

Backfill plan. Use the 2.4-billion-record history and the 105,000-record/s envelope. Define live reservation, borrow/reclaim, source/shuffle/state/sink token budgets, output identity, schema range, checkpoint interval, pause/abort thresholds, and comparison gates. Explain why a fixed 55,000-record/s backfill is unsafe during the 85,000-record/s live peak.

Late correction drill. A market partition is disconnected for three hours and returns events whose event times precede the final watermark. Decide which events correct existing windows, which enter reconciliation, how state is recovered or recomputed, how users see completeness, and how withdrawals differ from ordinary aggregates.

Effect-boundary review. A notification processor writes its consumer progress, calls an external provider, and stores a delivery record in three separate actions. Enumerate crash points and duplicate/loss outcomes. Redesign with a stable effect ID, durable state machine, provider idempotency/lookup where supported, and reconciliation. State the remaining limit honestly.

Durable rules for continuous systems

1. Define freshness from a named event boundary to a named visible result for a population.
2. Separate producer, gateway, log, processor, state, sink, and serving authority.
3. Choose partition keys from ordering, state locality, skew, and migration—not rate alone.
4. Measure lag in positions, bytes, ingest/event age, watermark delay, and visible age.
5. Size burst storage, processing recovery margin, scale delay, and sink capacity together.
6. Treat state, timers, windows, dedup identities, and checkpoint bytes as capacity.
7. Make watermarks and late-data correction semantics explicit; they are not wall clocks.
8. Join checkpoints to source positions, and prove restore plus replay from retained data.
9. Name the exact state/effect boundary before using an exactly-once claim.
10. Price repartitioning in bytes, state movement, skew, and rollback.
11. Reserve live capacity and make backfill borrowing interruptible across every constrained resource.
12. Test old schemas, old state, backfills, failures, and recovery—not only the current fast path.

Key-based systems concentrate design in identity and popularity; continuous systems add time, backlog, and movable state. Throughput-oriented analytical systems change the objective again: completion deadlines, scans, pruning, shuffle, skew, spill, and workload scheduling dominate. The same record and byte accounting will carry forward, but the unit of success becomes a completed query or batch rather than a continuously fresh result.

Evidence and transfer limits

• Apache Kafka 4.3’s current design documentation describes partitioned logs, consumer positions, replay, batching, replication, delivery semantics, and quotas. SignalWeave is product-neutral; verify the active Kafka version or another log’s acknowledgment, transaction, retention, and rebalance behavior before transferring claims.
• Akidau et al.’s Dataflow Model paper provides a primary model for event time, windowing, triggers, and the correctness/latency/cost trade. It is not evidence for SignalWeave’s fictional rates.
• Apache Flink’s current Watermark API documentation describes watermarks as stream progress indicators and explicitly allows heuristic late events. Framework APIs and combination/idleness semantics vary by version.
• Apache Flink 2.3’s current checkpoint operations documentation ties recovery to state and corresponding stream positions and describes checkpoint storage. It does not prove a recovery objective; measure the active job’s state, storage, compatibility, and restore path.
• Apache Avro 1.12.0’s specification defines concrete writer/reader schema resolution rules. Wire compatibility is not semantic compatibility, and other serialization systems use different rules.
• The deterministic fixture at examples/performance-engineering-system-design-handbook/part-05/event-streaming/ verifies the 300-second allocation, 72-million-record no-scale backlog, 80-minute drain, 2.4-million-record scaling backlog, 240-second catch-up, 1.026-TiB replicated one-hour burst, byte amplification, state/checkpoint sizes, and 44.44/12.12-hour backfill lower bounds. It assumes constant rates, average compressed size, one FIFO backlog, instant post-delay capacity, no skew, no control oscillation, and no sink/checkpoint interference. Replace it with partition distributions, stateful load tests, failure drills, and event-linked freshness evidence.