# Event Replay Low-Level Design: Replaying Historical Events for State Rebuild and Backfill Source: https://www.techinterview.org/post/3233468241/lld-event-replay/ Updated: 2026-04-18 · techinterview.org Event replay re-processes historical events through a consumer to rebuild state, backfill a new downstream system, fix a processing bug, or audit system behavior. It is a critical operational capability for event-sourced systems, CDC pipelines, and Kafka-based architectures. The design challenges are: accessing historical events (retention policy, storage), ensuring replay is idempotent (processing the same event twice produces the same result), and isolating replay traffic from live processing so that a replay job does not cause duplicate side effects in production systems. ## Core Data Model (Event Store) ``` CREATE TABLE Event ( event_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), aggregate_id VARCHAR(100) NOT NULL, -- the entity this event belongs to aggregate_type VARCHAR(50) NOT NULL, -- 'Order', 'User', 'Payment' event_type VARCHAR(100) NOT NULL, -- 'OrderPlaced', 'PaymentFailed' payload JSONB NOT NULL, metadata JSONB NOT NULL DEFAULT '{}', -- correlation_id, causation_id sequence_num BIGINT NOT NULL, -- monotonic per aggregate occurred_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), UNIQUE (aggregate_id, sequence_num) ); CREATE INDEX idx_event_aggregate ON Event(aggregate_id, sequence_num); CREATE INDEX idx_event_type_time ON Event(event_type, occurred_at); CREATE INDEX idx_event_occurred ON Event(occurred_at); -- for time-range replay -- Replay job tracking CREATE TABLE ReplayJob ( job_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), consumer_name VARCHAR(100) NOT NULL, replay_type VARCHAR(50) NOT NULL, -- 'full', 'aggregate', 'time_range', 'event_type' filter_params JSONB, -- {aggregate_id, from_time, to_time, event_types} status VARCHAR(20) NOT NULL DEFAULT 'pending', total_events BIGINT, processed_count BIGINT NOT NULL DEFAULT 0, last_event_id UUID, -- checkpoint for resume started_at TIMESTAMPTZ, completed_at TIMESTAMPTZ, created_at TIMESTAMPTZ DEFAULT NOW() ); ``` ## Idempotent Event Handler ``` class IdempotentEventHandler: """Wrap any event handler with idempotency tracking.""" def __init__(self, handler_name: str): self.handler_name = handler_name def handle(self, event: dict, is_replay: bool = False) -> bool: event_id = event['event_id'] idempotency_key = f"{self.handler_name}:{event_id}" # Check if already processed if redis.exists(f"processed:{idempotency_key}"): return False # skip — already handled # For DB-backed idempotency (survives Redis flush): existing = db.fetchone(""" SELECT 1 FROM ProcessedEvent WHERE handler_name=%s AND event_id=%s """, [self.handler_name, event_id]) if existing: redis.setex(f"processed:{idempotency_key}", 86400, 1) return False # Process the event with db.transaction(): self._process(event, is_replay=is_replay) db.execute(""" INSERT INTO ProcessedEvent (handler_name, event_id, processed_at) VALUES (%s, %s, NOW()) ON CONFLICT DO NOTHING """, [self.handler_name, event_id]) redis.setex(f"processed:{idempotency_key}", 86400, 1) return True def _process(self, event: dict, is_replay: bool = False): # Subclasses implement actual business logic here # is_replay flag can suppress side effects (emails, webhooks) during replay raise NotImplementedError ``` ## Replay Execution with Checkpointing ``` def run_replay_job(job_id: str, handler: IdempotentEventHandler, batch_size: int = 1000): job = db.fetchone("SELECT * FROM ReplayJob WHERE job_id=%s", [job_id]) db.execute("UPDATE ReplayJob SET status='running', started_at=NOW() WHERE job_id=%s", [job_id]) # Build query based on replay type filters = job['filter_params'] or {} where_clauses = ["1=1"] params = [] if filters.get('aggregate_id'): where_clauses.append("aggregate_id = %s") params.append(filters['aggregate_id']) if filters.get('from_time'): where_clauses.append("occurred_at >= %s") params.append(filters['from_time']) if filters.get('to_time'): where_clauses.append("occurred_at ( SELECT occurred_at FROM Event WHERE event_id=%s ) """) params.append(job['last_event_id']) query = f""" SELECT * FROM Event WHERE {' AND '.join(where_clauses)} ORDER BY occurred_at ASC, event_id ASC """ processed = 0 last_id = job['last_event_id'] for event in db.fetchall_cursor(query, params, batch_size=batch_size): handler.handle(event, is_replay=True) last_id = event['event_id'] processed += 1 # Checkpoint every 1000 events if processed % 1000 == 0: db.execute(""" UPDATE ReplayJob SET processed_count=%s, last_event_id=%s WHERE job_id=%s """, [processed, last_id, job_id]) db.execute(""" UPDATE ReplayJob SET status='completed', processed_count=%s, completed_at=NOW() WHERE job_id=%s """, [processed, job_id]) ``` ## Kafka-Based Replay For Kafka-backed systems, replay is built in: Kafka retains messages for a configurable retention period (days to forever with tiered storage). To replay, create a new consumer group and set its offset to the desired start position: ``` from kafka import KafkaConsumer, TopicPartition def replay_from_timestamp(topic: str, start_ms: int, consumer_group: str): consumer = KafkaConsumer(bootstrap_servers=['kafka:9092'], group_id=consumer_group, auto_offset_reset='earliest', enable_auto_commit=False) partitions = consumer.partitions_for_topic(topic) tps = [TopicPartition(topic, p) for p in partitions] consumer.assign(tps) # Seek all partitions to the timestamp offsets = consumer.offsets_for_times({tp: start_ms for tp in tps}) for tp, offset_and_ts in offsets.items(): if offset_and_ts: consumer.seek(tp, offset_and_ts.offset) for message in consumer: process_event(message.value, is_replay=True) consumer.commit() ``` ## Key Interview Points - Idempotency is the prerequisite for replay — if handlers are not idempotent, replaying events causes double-charges, duplicate emails, and corrupted state. Design for idempotency first. - The `is_replay=True` flag lets handlers suppress external side effects (sending emails, calling payment APIs) during replay — only update internal state. - Checkpointing makes replay resumable after failure — save the last processed event_id periodically so a crashed replay job restarts near where it left off, not from the beginning. - Replay in a separate consumer group (Kafka) or a separate handler instance to avoid interfering with live processing — different offsets, different DB connections, separate rate limits. - Kafka tiered storage enables infinite event retention at S3 cost (~$0.023/GB/month) vs local disk — essential for systems that need replay of events months or years old. - Sequence number per aggregate enables "rebuild this entity's state from scratch" by replaying only its events in order — the foundation of event sourcing. Event replay and data pipeline recovery design is discussed in [Databricks system design interview questions](https://www.techinterview.org/post/3233460280/databricks-interview-guide-2026-spark-internals-delta-lake-and-lakehouse-architecture/). Event replay and event sourcing system design is covered in [LinkedIn system design interview preparation](https://www.techinterview.org/post/3233460278/linkedin-interview-guide-2026-social-graph-engineering-feed-ranking-and-professional-network-scale/). Event replay and distributed system recovery design is discussed in [Uber system design interview guide](https://www.techinterview.org/post/3233460273/uber-interview-guide-2026-dispatch-systems-geospatial-algorithms-and-marketplace-engineering/). See also: [Scale AI Interview Guide 2026: Data Infrastructure, RLHF Pipelines, and ML Engineering](https://www.techinterview.org/post/3233460275/scale-ai-interview-guide-2026-data-infrastructure-rlhf-pipelines-and-ml-engineering/) See also: [Anthropic Interview Guide 2026: Process, Questions, and AI Safety](https://www.techinterview.org/?p=3233460282)