Distributed Systems Fundamentals Questions

Core principles and theory that underlie distributed computing systems. Includes understanding trade offs between consistency, availability, and partition tolerance, common consistency models such as eventual and strong consistency, replication and sharding strategies, load balancing and data partitioning, consensus algorithms and their guarantees, scalability and fault tolerance patterns, and how these concepts apply to infrastructure components such as databases, caches, service meshes, and load balancers. Candidates are expected to explain design choices, common failure modes, and how fundamental concepts influence architecture decisions for resilient and scalable systems.

MediumTechnical

75 practiced

Describe failure modes related to network partitions in cloud environments (e.g., cross-AZ latency spike vs full partition). For each failure type explain how distributed data systems typically behave (e.g., leader failover, split brain) and how you'd test and detect partition responses in a data platform.

Sample Answer

Start by categorizing common partition failure modes, describe typical distributed-system behavior, then give concrete tests and detection signals.

Failure modes and expected system behavior- Cross-AZ latency spike (increased RTT but packets still flow): systems see replication lag and slowed heartbeats. Leader-based systems (Raft/ZooKeeper) usually keep the leader but experience slower commit latencies; followers fall behind (increased replication lag) and client timeouts rise. No split-brain, but throughput and tail latency degrade.- Partial partition (some nodes cut off from subset of cluster): quorum-based systems may lose quorum on the minority side — that side becomes read-only or unavailable. If misconfigured fencing exists, leader stays with quorum side; misconfiguration can cause delayed leader election or write rejection.- Full partition (network split into disjoint groups): risk of split-brain if majority rules not enforced or if administrative mistakes allow dual primaries. Eventually each side may elect leaders independently causing divergent writes and hard reconciliation.- Asymmetric partition (one-way connectivity): leads to confusing behavior (nodes appear alive to one side, dead to the other) and can break leader heartbeats causing flapping elections.- Flapping/intermittent partitions: repeated leader elections, state churn, increased GC/CPU from retries, complicates consistency guarantees.

How distributed data systems typically behave (examples)- Raft/Paxos: require majority quorum. On minority partitioned side, no leader elected; writes blocked until quorum restored. On majority side, leader continues; once healed, followers catch up via replication logs.- Leaderless systems (Cassandra): prefer configurable consistency levels. With partitions, writes may succeed on isolated nodes depending on CL, causing divergent replicas and requiring read-repair/anti-entropy to converge later.- Systems with weak fencing or manual failover: risk split-brain (two actives) and conflicting writes.

Testing and fault-injection strategies- Controlled chaos engineering: run fault-injection in staging with tools that simulate loss/latency.- Network shaping examples (Linux tc):

bash

# add 200ms latency and 5% packet loss to eth0
sudo tc qdisc add dev eth0 root netem delay 200ms loss 5%
# remove
sudo tc qdisc del dev eth0 root

- Use toxiproxy or Linkerd fault-injection for per-service simulation; use cloud provider features to simulate AZ blackhole (wreck traffic tables) or use iptables to drop traffic to specific CIDRs.- Simulate one-way partitions: block outgoing but not incoming to reproduce asymmetric cases.- Inject high-latency DNS responses to simulate control-plane delays.

Detection and observability- Key metrics to monitor: - Leader election count and time-to-election - Replication lag and last-applied indexes per replica - Request latencies and error rates (writes rejected, timeouts) - Node reachability and TCP connection churn - SLO/SLA violations and synthetic canary test outcomes- Tracing/logging: distributed traces to see increased span durations and hotspots; leader/follower logs to detect state transitions and split-brain indicators (concurrent leader logs).- Alerts: - Spike in leader elections > threshold - Replication lag > acceptable window - Increase in write errors or decreased successful commit rate - Node count in cluster not matching expected for > N minutes- Automated canaries: periodic writes with read-after-write checks at different consistency levels to detect stale reads or divergence.

Mitigations and test assertions- Ensure quorum-configured safety and implement fencing (e.g., epoch/tokens) to prevent split-brain.- Test that system blocks unsafe writes during minority partitions and recovers without data loss when healed.- Validate backoff/retry behavior to avoid cascading load during flaps.- For leaderless systems, verify anti-entropy processes converge within SLAs and that hinted-handoff capacity is bounded.

Summarize test plan (practical steps)1. Baseline metrics and synthetic canaries.2. Inject cross-AZ latency; assert increased latencies but no split-brain; measure failover time.3. Inject partial partition isolating minority; assert minority rejects writes, majority serves, and reconciliation occurs post-heal.4. Inject asymmetric block to force flapping; assert leader stability and alerting.5. Run chaos for extended period and verify SLOs, alert fidelity, and recovery automation work.

This approach ensures you understand failure modes, how systems behave under each, and have repeatable tests and observability to detect and harden the platform.

HardSystem Design

85 practiced

Design a distributed deduplication service for streaming events that must detect duplicates across partitions and consumers with low latency, support a configurable deduplication TTL window, and scale to millions of ops/sec. Describe data structures, sharding strategy, memory/disk trade-offs, and recovery after failures.

Sample Answer

Requirements (clarify): dedupe streaming events across partitions & consumers, low latency (<ms–tens ms), configurable TTL window (seconds → hours), throughput millions ops/sec, scalable, fault-tolerant, near exactly-once detection.

High-level architecture:- Ingest → Partitioned stream (Kafka/Kinesis) → Deduplication layer (sharded state stores) → Downstream consumers.- Dedup layer implemented as horizontally sharded stateless frontends + persistent state backends (in-memory index + persistent log/rocksdb) per shard.

Sharding strategy:- Shard by event dedupe key hash (consistent hashing) so identical keys route to same shard. Use enough shards to scale; support resharding via consistent hashing + state migration.

Data structures:- Each shard maintains a time-indexed hash table: key → (last_seen_ts, metadata). Implement as: - In-memory LRU/time-wheel for TTL eviction (e.g., ring buffer buckets for each 1s slot). - Persistent store (RocksDB) for cold entries and recovery: key → last_seen_ts.- For very high throughput, use a Bloom filter front-layer to reject obvious non-duplicates cheaply, then consult hash table for hits to avoid false positives.

Memory vs disk trade-offs:- Keep hot window (TTL window) fully in RAM for O(1) checks and low latency. Evict older entries to RocksDB to bound memory. RocksDB gives persistence and compaction; a write-ahead commit log (append-only) can help fast restore.- Bloom filter reduces RAM by sacrificing small false-positive rate; tune based on acceptable duplicate suppression risk.

Processing flow:1. Event arrives → route to shard.2. Shard checks bloom filter; if negative, accept and insert.3. If positive, lookup in-memory hash: if last_seen_ts within TTL → mark duplicate/drop; else accept and update timestamps (write to in-memory + write-ahead log + RocksDB async).4. Write results to downstream.

Scaling:- Autoscale frontends independently; scale shards by adding nodes and migrating ranges. Use partition rebalancer that streams state via snapshot + delta WAL.

Consistency & exactly-once considerations:- Use idempotent writes and sequence numbers. Dedup decision is deterministic per key+timestamp. Persist decision via synchronous WAL before acknowledging if exact durability required; otherwise async for latency.

Failure & recovery:- On shard crash: replay WAL + restore RocksDB snapshot; rebuild in-memory window by scanning keys with last_seen_ts within TTL (can maintain a compacted in-memory snapshot or time-indexed checkpoint every N seconds). Use incremental snapshots to reduce recovery time.- For resharding, perform consistent migration: freeze routing for key range, snapshot state, transfer snapshot, apply WAL deltas, switch routing.

Operational concerns:- Monitor false-positive rate, memory pressure, tail latency.- TTL accuracy: use monotonic clocks and allow clock skew handling (source timestamps vs ingestion).- Edge cases: very high-cardinality keys → consider sampling or backpressure; long TTLs → high memory/disk needs; out-of-order events → use event time with allowed lateness.

This design balances low-latency in-memory checks with durable disk-backed state, supports configurable TTL, scales via consistent hashing, and provides fast recovery via WAL + snapshots.

MediumTechnical

83 practiced

Explain how serialization formats (JSON, Avro, Protobuf, Parquet) affect distributed system behavior: payload size, schema evolution, deserialization CPU, and network throughput. As a data engineer, how do you choose formats for different pipeline segments (ingest, intermediate transport, storage)?

Sample Answer

High-level summary: serialization format choice impacts wire size, CPU for (de)serialization, ability to evolve schemas safely, and thus throughput/latency. Trade-offs matter across pipeline stages (ingest, transport, storage).

How formats compare:- JSON: human-readable, schema-less. Payload size: large (text). Deserialization CPU: moderate to high (parsing strings). Schema evolution: ad-hoc — no enforced compatibility. Network throughput: lower due to size + parsing cost. Good for ad-hoc ingest or external APIs.- Avro: compact binary, schema stored separately or with payload. Payload size: small. Deserialization CPU: low–moderate with codegen. Schema evolution: first-class (back/forward compatible with rules). Network throughput: high. Great for streaming, Kafka.- Protobuf: very compact, efficient codegen, strong typing. Payload size: very small. Deserialization CPU: low. Schema evolution: supported with constraints (field numbers, reserved). Network throughput: very high. Good for RPC and high-throughput streams.- Parquet: columnar file format, optimized for analytics. Payload size: small (column compression, encoding). Deserialization CPU: higher for nested types but efficient for column reads. Schema evolution: supported (add columns), but more complex. Network throughput: optimized for bulk reads, not per-record transport. Best for storage and batch processing.

Choosing per pipeline segment:- Ingest (edge/API): use JSON or Protobuf depending on clients. If clients are heterogeneous and human-readable matters, JSON. If you control producers and need throughput, Protobuf.- Intermediate transport (streaming, messaging): use Avro or Protobuf. Avro works well when schema registry is desired (Kafka + Confluent). Protobuf if minimal size and fast RPCs matter.- Storage (data lake/warehouse): use Parquet (or ORC) for analytics: columnar compression reduces storage and improves query I/O. Keep original Avro/Protobuf for replayability if you need record-level fidelity.Operational considerations:- Use schema registry (Avro/Protobuf) to manage evolution and compatibility checks.- Measure end-to-end latency and CPU: compact formats reduce network but may increase CPU if complex (e.g., nested Protobuf). Benchmark with representative payloads.- Balance developer ergonomics vs operational needs: JSON for flexibility, Parquet for analytics, Avro/Protobuf for robust streaming.

Example pattern: API -> ingest layer accepts JSON; ingesters convert to Avro with schema registry and write to Kafka; stream processing operates on Avro/Protobuf; final processed datasets written to Parquet in S3 for analytics.

HardTechnical

70 practiced

Discuss how data locality impacts scheduling in big data clusters (e.g., Spark on YARN). How does locality affect shuffle costs, task throughput, and overall job latency? Propose strategies to improve locality in a multi-tenant cluster and trade-offs for fairness vs locality.

Sample Answer

Data locality strongly influences scheduling in big-data clusters because moving computation to where data lives minimizes network I/O, reduces shuffle cost, and improves task throughput and job latency.

Impact:- Shuffle costs: Remote reads increase cross-rack or cross-datacenter transfers during shuffle and spill phases, raising network congestion and disk I/O; more retries and GC from longer task runtime increase shuffle overhead.- Task throughput: Local tasks finish faster and free executors sooner, enabling higher task parallelism. Remote tasks are slower and reduce effective throughput.- Job latency: Poor locality lengthens individual task durations and increases stragglers, extending stage and job completion time—especially bad for small-latency pipelines.

Strategies to improve locality (multi-tenant):1. Data-aware scheduler: Use YARN/Spark locality levels (PROCESS, NODE, RACK) and prefer node-local tasks; increase node-local allocation timeouts slightly to wait for local slots.2. Data placement policies: Co-locate hot datasets on more nodes (replication), use HDFS rack-aware placement to favor tenants’ compute clusters.3. Cache hot datasets: Use in-memory caching (Spark persist) or distributed cache layers (Alluxio) for repeated reads.4. Dynamic capacity/pools: Reserve capacity (YARN queues or K8s node pools) for tenants with data-local needs; use preemption sparingly.5. Job packing & affinity: Schedule jobs with affinity to nodes holding their partitions; use container placement constraints.6. Network improvements: Upgrade to higher bandwidth, enable RDMA or improved shuffle services (e.g., Spark’s external shuffle service) to reduce remote-read penalty.

Trade-offs (fairness vs locality):- Favoring locality (waiting for local slots, replication) can reduce fairness and increase wait times for other tenants; replication increases storage cost.- Enforcing strict fairness (FIFO/strict capacity) can hurt locality and increase job latency.Balanced approaches:- Soft priorities: allow short locality wait time before broader scheduling.- Weighted fair-sharing with locality boost: give higher weight to node-local allocations but cap per-tenant usage.- Adaptive policies: monitor cluster metrics (network, queue wait, task slowdown) and dynamically tune locality wait thresholds or replicate hot partitions temporarily.

In practice: start with moderate locality wait (e.g., 3s), enable caching for hot jobs, replicate hot datasets selectively, and implement locality-aware fair sharing. Measure end-to-end job latency, network utilization, and fairness SLAs to tune trade-offs.

EasyTechnical

63 practiced

Explain load balancing fundamentals relevant to data ingestion and microservices: differences between L4 (TCP) and L7 (HTTP) load balancing, sticky sessions (session affinity), and client-side vs server-side load balancing. How do these choices affect stateful vs stateless services in a distributed data pipeline?

Sample Answer

Brief answer: Load balancing directs traffic across multiple service instances to improve throughput, availability and latency. For data ingestion and microservices you choose L4 vs L7, session affinity, and client- vs server-side balancing based on protocol, observability, and whether services are stateful.

L4 (TCP) vs L7 (HTTP):- L4 operates at transport layer (IP/TCP): fast, low overhead, routes by IP/port; can't inspect HTTP headers or do content-based routing. Good for raw TCP streams (Kafka ingestion, TLS passthrough).- L7 operates at application layer (HTTP/REST/gRPC): can route by URL, headers, cookies, implement retries, circuit breaking, path-based routing, and observability (metrics/tracing). Ideal for REST microservices and HTTP-based ingestion endpoints.

Sticky sessions (session affinity):- Binds client requests to the same backend (via cookie, source IP). Useful when backend keeps in-memory session/state.- For data pipelines prefer avoiding affinity: it reduces load distribution and hampers scaling. If needed, limit scope and use short-lived affinity or move state to external store (Redis, DB) to preserve stateless backends.

Client-side vs server-side:- Server-side (reverse proxy/load balancer like HAProxy, ALB, Envoy) centralizes routing, health checks, and L7 features.- Client-side (service discovery + client load balancing, e.g., Kafka producers, gRPC clients + Consul) gives lower latency and fine-grained control, but pushes complexity to clients.- For data ingestion at scale, combine both: server-side ingress for secure, uniform entry; client-side for service-to-service calls within cluster.

Impact on stateful vs stateless:- Stateless services pair well with any balancer; they scale horizontally and enable retries.- Stateful services require affinity or externalized state. Best practice: externalize state (object store, DB, cache) so services remain stateless and you can use L7 with autoscaling, observability, retries, and safer failure handling.

Example: Use an ALB (L7) for HTTP ingestion endpoints with JSON validation and rate limiting, migrate session data to Redis, and use client-side discovery for high-throughput inter-service streaming (Kafka producers) to minimize hop count.

Unlock Full Question Bank

Get access to hundreds of Distributed Systems Fundamentals interview questions and detailed answers.

Join thousands of developers preparing for their dream job.