Real-Time and Streaming System Design Questions

Requirements & constraints:

• Ingest 200k events/s, downstream steady capacity 50k/s. Need durable buffering, predictable recovery, and upstream flow control.

Architecture & queueing:

• Primary buffer: Kafka topic(s) — durable, partitioned, high-throughput, configurable retention. Kafka lets producers continue writing while consumers lag, supports replay, and is fault-tolerant.
• Short-term in-memory buffer at the ingestion gateway (bounded queue per producer connection) to absorb microbursts and apply backpressure quickly. Limit size (e.g., a few seconds of events) to avoid OOM.

Partitioning & parallelism:

• Partition Kafka by a high-cardinality key that preserves processing locality where needed (e.g., user_id, tenant). Create N partitions so you can run M consumer instances in a consumer group to parallelize processing. Choose partitions >= max parallel consumers you may scale to.
• Consumer instances should batch reads (size tunable) to increase downstream throughput and reduce per-message overhead.

Throttling vs dropping:

• Prefer throttling when possible: expose rate-limit responses (HTTP 429 + Retry-After) or a token-bucket-based pushback protocol so well-behaved clients slow down.
• If upstream cannot be trusted or latency/SLA constraints exist, use policy tiers:
  • Tier A (critical): always retained.
  • Tier B (best-effort): if backlog > threshold, sample or drop with metrics.
• Dropping policies: drop oldest from best-effort pool or probabilistic sampling; always record drop metrics and optionally emit a skip-marker event so consumers know data missing.

Signaling upstream:

• For HTTP clients: 429 with Retry-After, and include current ingestion lag and suggested backoff.
• For persistent connections: implement a credits/token protocol or pause reads on TCP; or use gRPC flow-control / windowing.
• For Kafka producers: use broker-side quotas and throttle producers; or return explicit rate limit errors from the ingestion gateway.

Recovery / catch-up plan:

• Maintain consumer lag metrics; auto-scale consumers (increase instances, increase CPU/IO) up to downstream constraints.
• If downstream is the bottleneck, use burst-processing: temporarily increase batching and parallelism, but respect downstream capacity (avoid overload).
• Use prioritized catch-up: consume critical-topic partitions first. For best-effort backlog, process at lower priority or during off-peak.
• If sustained overload: extend Kafka retention, persist backlog, notify stakeholders, and allow offline reprocessing (replay from offsets). Implement circuit-breaker to refuse new low-priority writes when disk/retention risk high.
• Ensure idempotent consumers and checkpoints to allow safe replay.

Monitoring & safety:

• Track ingestion rate, consumer throughput, partition lag, retention usage, drop/Throttled counts.
• Alerts for sustained lag, retention nearing capacity, or repeated 429s.
• Add SLA-based autoscaling, and run periodic chaos tests to validate backpressure behavior.

Trade-offs:

• Durable (Kafka) = reliability, replayability, higher cost; in-memory = low-latency but volatile. Use both: in-memory for immediate flow control, Kafka for durable buffering and recovery.

Requirements clarification: reliably ingest 1,000,000 events/sec peak, deliver to analytics stores with <5s tail latency, cloud-hosted, durable, partitioned, fault-tolerant.

High-level design:

• Edge / Ingress: autoscaled HTTP/gRPC collectors behind LB (ALB/GCLB) or managed Kafka producers (Kafka REST Proxy). Use client libraries that support batching, compression, and retries.
• Durable buffer: distributed log (Apache Kafka / AWS MSK or Confluent, or Google Pub/Sub / AWS Kinesis). This provides durable, ordered, partitioned storage and decouples producers from consumers.
• Partitioning: choose partition key based on downstream access pattern (user_id, device_id). Ensure hot-key mitigation (hash+salt, consistent hashing, or dynamic re-sharding). Right-size partitions: capacity calc example below.
• Consumers: stream processing layer (Kafka Streams / Flink / Spark Structured Streaming / Dataflow) that reads from partitions, performs lightweight enrichment/validation, writes to analytics stores (OLAP DB, S3, warehouse).
• Downstream sinks: durable writes to object storage (S3/GS), and low-latency stores (ClickHouse/BQ/Redshift) via micro-batches or streaming connectors.

Sizing example:

• If avg event = 1 KB -> 1 GB/s (~8 Gbps). For Kinesis (1 MB/s per shard) → ~1000 shards. For Kafka with 200 MB/s per broker and 5 brokers you get ~1000 MB/s; plan partitions ~200–1000 depending on throughput and parallelism. Use benchmarks with your payload to finalise.

Buffering & latency:

• Producers batch (e.g., 100–1000 events or 50–200ms max) and compress to reduce traffic.
• Keep stream retention short-term (hours) for replay; use long-term sink to S3 for archival.
• Consumers process in small micro-batches (100–500ms) or event-by-event if using Flink to meet <5s tail latency.
• Use backpressure: if consumers lag, scale consumers horizontally; if impossible, apply shedding policies or route to cold storage while alerting.

Ordering & guarantees:

• Provide per-partition ordering; pick keys to scope ordering to relevant entity.
• Use at-least-once delivery with idempotent sink writes or exactly-once semantics if stream engine supports it (Kafka transactions, Flink two-phase commits). Prefer idempotent writes (dedupe via event id) for sinks like S3/warehouse.
• For global ordering needs, funnel through a single partition (costly) — avoid if high throughput required.

Failure & retry strategy:

• Producers: client retries with exponential backoff and bounded queue; circuit-breaker to avoid cascading failures.
• Broker layer: multi-AZ cluster, replication factor >=3, enable controller quorum, monitor ISR.
• Consumers: checkpoint offsets atomically after successful sink commit. On consumer failure, reprocess from last checkpoint.
• Retries: ephemeral automatic retries in stream processor; persistent failures send to DLQ (Kafka topic or S3) with failure metadata for offline handling.
• Disaster recovery: cross-region replication (MirrorMaker / MSK cross-region replication or Pub/Sub replication) and periodic snapshots to S3.
• Monitoring & SLOs: track end-to-end latency percentiles, consumer lag, throughput, error rates; alert on >1s lag growth or partition hot-spot. Use autoscaling policies based on lag + CPU.

Trade-offs:

• Kafka/MSK gives high throughput and fine-grained control; managed Pub/Sub/Kinesis simpler operationally but may require more shards for small events.
• Exactly-once semantics increase complexity; idempotence + dedupe often simpler and robust.

Why this works:

• Durable, partitioned log decouples producers/consumers for reliability.
• Batching + compression reduces network and broker load; micro-batching in processors keeps tail latency low.
• Per-key partitioning preserves ordering where needed while allowing massive horizontal parallelism and autoscaling to handle 1M eps.

Requirements & constraints:

• Exactly-once end-to-end from producers to analytical store despite retries, failures, duplicates, and scale.
• Low-latency streaming pipeline: producers → message broker → stream processor → analytical store.
• Assume use of mature components (e.g., Kafka, Flink/Beam, transactional DB or OLAP store supporting atomic writes).

Architecture (high level):

• Producers: implement idempotent producers with per-producer monotonically increasing sequence numbers or unique event IDs (UUID + logical timestamp). Retry-safe client library that preserves sequence/IDs across retries.
• Broker: use a broker with producer idempotence and transactions (Kafka’s ProducerId + transactions). Group related writes into transactional batches so broker will expose either committed or aborted sequences. Broker maintains offsets atomically per partition.
• Stream processor: use a streaming engine with exactly-once processing semantics via checkpointing and two-phase commit (e.g., Flink with checkpoint barriers + XA/Transaction sink or two-phase commit sink). Processor reads committed broker records, maintains state with durable, periodic checkpoints (offsets + operator state) to a reliable store (e.g., S3 or cluster metadata store). On checkpoint completion, sink commit is triggered atomically.
• Sink / Analytical store: sink must support transactional/atomic commits or idempotent upserts keyed by event ID. Options:
  • Use transactional writes: write staged files/rows and commit atomically (e.g., Iceberg/Hudi/Delta Lake + metadata commit).
  • Or use an append with deduplication using the event ID with a unique constraint (requires transactional DB or upsert semantics).

Mechanisms & flow:

1. Producer sends message with producer-id + seq / event-id. Broker accepts and stores under a transaction.
2. Processor consumes only committed broker transactions, applies transformations updating in-memory state.
3. On checkpoint, processor atomically:
   • flushes sink writes to a staging area,
   • writes checkpoint metadata including broker offsets and sink transaction id,
   • only upon successful checkpoint completion, atomically commits sink transaction (two-phase commit).
4. On restart, processor restores state from checkpoint and either replays or finalizes pending sink transactions ensuring no duplicate commit.

Performance & complexity trade-offs:

• True exactly-once incurs overhead: transactional protocol latency, larger metadata writes during checkpoints, and potential backpressure.
• Checkpoint frequency: shorter intervals reduce potential duplicate reprocessing window but increase IO and CPU load.
• Two-phase commit sinks add RTT and coordination with external store; using storage engines with native atomic commit (Iceberg/Delta) reduces complexity vs distributed XA.
• Idempotent producers add minor client-side complexity but low runtime cost.

Failure modes & mitigation:

• Mid-checkpoint crash: processor will recover and re-execute since sink commit is gated on checkpoint; use transactional staging to avoid partial commits.
• Broker unavailability: producers buffer or use durable local persist; monitor leader loss and partition rebalances.
• Sink partial commit: use idempotent commit metadata + atomic rename/metadata update to avoid visibility of partial data.

Testing plan (end-to-end correctness):

1. Unit tests: producer client idempotence, processor stateful ops, sink atomic commit logic.
2. Integration tests with chaos:
   • Inject network partitions between components.
   • Kill processors during checkpoint window; assert no duplicates and state correctness.
   • Broker failover while producing/consuming.
   • Sink failures during commit; ensure either committed once or rolled back.
3. Property tests: generate randomized event streams with duplicates, out-of-order, retries; assert final analytical store deduplicated and matches expected aggregations.
4. End-to-end deterministic replay: run small workload, record expected result, run pipeline with injected failures, compare.
5. Load & latency tests: measure throughput vs checkpoint interval, measure commit latency; tune checkpoint frequency and batching.
6. Observability checks: end-to-end lineage tracing (event IDs), metrics for committed transactions, consumer offsets, checkpoint durations. Implement alerting on mismatch counters.

Summary:
Use idempotent producers + transactional broker + stateful processor with checkpoint-coordinated two-phase commits to sink (or native atomic commit storage). This yields strong correctness at cost of extra coordination, IO, and careful testing/observability.

Requirements & constraints:

• Ingest 200k events/sec total across 3 regions (~66k/sec/region peak).
• Deduplicate by event_id within a configurable time window (e.g., 24h).
• Bounded memory, minimal false negatives (prefer false positives over negatives).
• Low added latency, high availability, graceful recovery.

High-level design:

• Regional frontends receive events, do fast local dedupe, then forward canonical events to global sinks (analytics/DB).
• Two-layer dedupe: fast probabilistic per-region + durable exact windowed store.

Per-region layer (fast path):

• Use a time-partitioned Bloom filter ring (e.g., one filter per 5-minute slot; rotate). Insert event_id on first-seen; if Bloom says seen => treat as duplicate locally.
• Bloom tuned for low false-negative (Bloom has zero false negatives) and acceptable false positives; size chosen per slot using expected QPS and retention.
• This gives extremely cheap memory and CPU for initial filtering and reduces cross-region noise.

Durable windowed store (exact):

• For correctness, regionally write event_id to an eventually-consistent key-value store with TTL equal to dedupe window (e.g., DynamoDB/Bigtable/Cassandra) on first pass; reads check existence to ensure exact dedupe when Bloom is ambiguous.
• To bound storage, use sharded keys and TTL compaction; per-region storage handles only that region's writes.

Cross-region coordination & global dedupe:

• To remove duplicates across regions, use an idempotent global sink and an ingestion stream with exactly-once semantics where possible:
  • Each region forwards "candidate" events to a global Kafka topic partitioned by event_id hash; consumers are a small set of dedupe workers that maintain a windowed RocksDB store (local LRU + persistent SSTables) keyed by event_id with timestamps.
  • Partitioning by event_id guarantees all events with same id go to same worker; worker checks RocksDB for existing id within window; if absent, accept and write id with TTL; if present, drop.
• This ensures global single-authority per event_id without synchronous cross-region calls.

Eventual reconciliation:

• Since regional Bloom filters can false-positive (drop unique) but never false-negative, the bigger risk is cross-region races and outages. To minimize false negatives:
  • Ensure workers use persistent stores with tombstones and replication factor >1.
  • Periodic reconciliation job: compact and scan global storage to detect missing accepted events by comparing raw ingestion logs (S3/GCS) vs accepted outputs; reprocess missing ones.
  • Use watermarking to bound reconciliation window.

Failure & recovery:

• Partial outage of a region: region can continue local Bloom filtering and buffer forwarded events in durable queue (local Kafka). Once recovered, replay preserved events to global topic; global dedupe workers will handle duplicates idempotently.
• If a dedupe worker fails: rely on Kafka consumer group rebalancing and RocksDB state snapshards persisted to shared storage or use changelog Kafka topics to reconstruct local state.
• To avoid losing dedupe state, use stream-processing frameworks (Flink/Kafka Streams) with checkpointing and state backend (RocksDB backed by durable storage). This provides fast recovery and consistent window semantics.

Trade-offs:

• Bloom filters reduce memory but cause false positives (drop unique events). Mitigate by using Bloom as fast hint + exact store for critical events, or use Counting Bloom / Cuckoo filter to allow deletes on rotation.
• Synchronous cross-region dedupe would be slow; partitioned global authority balances consistency and latency.
• Storage costs for exact window scale with QPS * window size; tune slot durations and compression.

Metrics & monitoring:

• Track false-positive rate, latency from ingest to sink, backlog size per region, reconciliation deltas, and state persistence lag.
• Alert on backlog growth, worker restarts, or reconciliation mismatches.

Why this meets constraints:

• Bounded memory via time-partitioned Bloom filters and TTLed storage.
• Minimal false negatives by using an exact persistent global dedupe layer and strong stream-processing state management.
• High throughput via sharding by event_id and local fast-path filtering; resilient to outages via durable buffering and checkpointed state.

Idempotency means performing the same operation multiple times has the same effect as doing it once. For data ingestion APIs this prevents duplicates and inconsistent state when clients retry due to network failures or timeouts — critical for exactly-once or deduplicated ingestion at scale.

Two practical patterns:

1. Client-provided idempotency keys

• How: Client generates a unique id (UUID, hash of record) per logical write and submits it with the request. Server stores key → outcome (accepted/processed) and deduplicates repeated requests by key.
• Trade-offs:
  • Storage: requires durable index/table of recent keys and outcomes (TTL window or longer if needed).
  • Latency: lookups add a small read before write; can be optimized with in-memory cache.
  • Complexity: medium — need key lifecycle, collision handling, and API semantics.
• GC/Cleanup: enforce TTL or retention policy; purge old keys after safe processing window to bound storage.

2. Deterministic object naming (write with unique key, e.g., S3)

• How: Client deterministically names the target (e.g., hash of payload + timestamp bucket) and writes idempotently (PUT with same key overwrites or uses conditional put like S3’s If-None-Match to avoid overwrite).
• Trade-offs:
  • Storage: storage grows with unique objects unless overwrite/compaction used; deduplication can be handled during downstream compaction.
  • Latency: usually low — direct write to object store; metadata-free if name encodes uniqueness.
  • Complexity: lower on server side; complexity shifts to naming scheme and downstream dedupe/compaction.
• GC/Cleanup: require lifecycle rules (S3 lifecycle policies), periodic compaction jobs to merge/expire duplicates, and retention policies to reclaim space.

Summary: client keys give stronger server-side control and easy immediate dedupe but cost metadata storage and lookup latency; deterministic naming is simpler and scales well with object stores but needs thoughtful naming, downstream dedupe, and lifecycle/compaction to control storage. Choose based on SLAs, scale, and where complexity should live (client vs platform).