This exercise builds a complete end-to-end Kafka pipeline for IPL live match events using a local Kafka cluster simulated with the `kafka-python` library against a containerised Kafka instance. You will implement a durable producer that publishes delivery events with Avro serialisation, a consumer group with at-least-once delivery semantics and manual offset commits, a dead-letter queue (DLQ) for unprocessable messages, and a monitoring function that reports consumer lag. All components are tested against explicit correctness assertions before the exercise is considered complete.
The exercise uses a mock Kafka infrastructure class that replaces the actual Kafka cluster with an in-memory implementation, enabling the full pipeline logic to be exercised and tested without requiring a running Kafka cluster. The mock faithfully reproduces Kafka's key behaviours: partition-key-based routing, ordered delivery within a partition, consumer group offset tracking, and at-least-once re-delivery on simulated consumer failure. All assertions run against this in-memory mock.
Analogy🏏Cricket
🏏 Think of it like cricket: Imagine the DRS review system deployed across three independent video-review centres in Mumbai, Chennai, and London, each holding a copy of the ball-tracking data. A CAP partition is a network outage that cuts communication between them. A CP system says: if the centres cannot synchronise, no review decision is issued — no player walks until the system is restored. Consistency is guaranteed; availability is sacrificed. An AP system says: each centre issues its own decision based on its local data, even if that means Mumbai says out and London says not out — reviews continue but different centres may give different verdicts. Partition tolerance is non-negotiable because the network always has the possibility of failing; the choice is whether umpires wait for consensus or proceed with local data.
Step 1 — Mock Kafka Infrastructure and Producer
Implement the `MockKafka` class that provides in-memory topic storage with partition-key routing, and the `IPLEventProducer` class that serialises delivery events and publishes them with correct partition key assignment. Produce 60 delivery events across three matches with a 2-second simulated match — verify that all events for the same bowler-match combination land in the same partition, confirming partition-key routing correctness.
Analogy🏏Cricket
🏏 Think of it like cricket: the MockKafka setup is a full-rules net session before match day — the same laws apply as in the middle, but in an environment you control. Partition-key routing is assigning each bowler-match pairing its own net lane: every delivery Bumrah bowls in match one goes down lane three, in strict bowling order, so the analyst watching that lane sees his spell exactly as it unfolded — ball one before ball two, never interleaved with another bowler's deliveries. That is precisely Kafka's guarantee: same key, same partition, order preserved within it. Verifying that all 60 events for each bowler-match combination landed in one partition is the coach walking the lanes to confirm no delivery strayed into the wrong net. Just as a spell only makes sense watched in sequence — the bouncer sets up the yorker — downstream consumers can only reason about an event stream whose per-key ordering is guaranteed. Prove the routing in the nets, and match day inherits the guarantee.
python
# exercise_kafka_pipeline.py — Step 1: Mock Kafka and Producer
import json
import hashlib
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional
import random
random.seed(42)
# ── Mock Kafka Infrastructure ─────────────────────────────────────────────────
@dataclass
class MockMessage:
topic: str
partition: int
offset: int
key: Optional[str]
value: dict
class MockKafka:
"""
In-memory Kafka cluster mock.
Supports: partition-key routing, ordered delivery per partition,
consumer group offset tracking.
"""
N_PARTITIONS = 4
def __init__(self):
# topic → partition → list of messages
self.topics: dict[str, dict[int, list[MockMessage]]] = defaultdict(
lambda: defaultdict(list)
)
# group_id → topic → partition → committed_offset
self.committed: dict[str, dict[str, dict[int, int]]] = defaultdict(
lambda: defaultdict(lambda: defaultdict(int))
)
def produce(self, topic: str, key: Optional[str], value: dict) -> MockMessage:
partition = self._partition_for_key(key)
offset = len(self.topics[topic][partition])
msg = MockMessage(topic, partition, offset, key, value)
self.topics[topic][partition].append(msg)
return msg
def _partition_for_key(self, key: Optional[str]) -> int:
if key is None:
return random.randint(0, self.N_PARTITIONS - 1)
return int(hashlib.md5(key.encode()).hexdigest(), 16) % self.N_PARTITIONS
def consume_batch(
self, topic: str, group_id: str, max_records: int = 100
) -> list[MockMessage]:
batch = []
for p in range(self.N_PARTITIONS):
committed_offset = self.committed[group_id][topic][p]
messages = self.topics[topic][p][committed_offset:committed_offset+max_records]
batch.extend(messages)
if len(batch) >= max_records:
break
return batch
def commit(self, group_id: str, messages: list[MockMessage]) -> None:
for msg in messages:
current = self.committed[group_id][msg.topic][msg.partition]
self.committed[group_id][msg.topic][msg.partition] = max(current, msg.offset + 1)
def lag(self, topic: str, group_id: str) -> dict[int, int]:
return {
p: len(self.topics[topic][p]) - self.committed[group_id][topic][p]
for p in range(self.N_PARTITIONS)
}
# ── Producer ──────────────────────────────────────────────────────────────────
class IPLEventProducer:
def __init__(self, kafka: MockKafka, topic: str = "ipl-deliveries"):
self.kafka = kafka
self.topic = topic
self.produced = 0
def publish_delivery(self, event: dict) -> MockMessage:
key = f"{event['match_id']}:{event['bowler']}"
msg = self.kafka.produce(self.topic, key, event)
self.produced += 1
return msg
# ── Generate IPL delivery events ──────────────────────────────────────────────
mock_kafka = MockKafka()
producer = IPLEventProducer(mock_kafka)
BOWLERS = ["Bumrah", "Shami", "Hardik"]
MATCHES = [10001, 10002, 10003]
event_id = 1
for match_id in MATCHES:
for bowler in BOWLERS:
for ball in range(1, 7): # 6 deliveries per bowler per simulated over
event = {
"delivery_id": event_id,
"match_id": match_id,
"over": 1,
"ball": ball,
"bowler": bowler,
"batter": random.choice(["Rohit","Kohli","Gill"]),
"runs": random.choice([0,1,2,4,6]),
"is_wicket": random.random() < 0.05,
"ts": datetime.now(timezone.utc).isoformat(),
}
producer.publish_delivery(event)
event_id += 1
print(f"Produced {producer.produced} events across {len(MATCHES)} matches")
# Verify partition-key routing: all deliveries for same bowler+match → same partition
for match_id in MATCHES:
for bowler in BOWLERS:
key = f"{match_id}:{bowler}"
expected_partition = mock_kafka._partition_for_key(key)
actual_partitions = set(
msg.partition
for p in range(mock_kafka.N_PARTITIONS)
for msg in mock_kafka.topics["ipl-deliveries"][p]
if msg.key == key
)
assert actual_partitions == {expected_partition}, \
f"Key {key} routed to multiple partitions: {actual_partitions}"
print("Partition routing verified ✓")Step 2 — Consumer with DLQ and Failure Simulation
Implement the `IPLEventConsumer` that processes batches of delivery events, routes malformed events to a dead-letter queue topic, commits offsets only after successful processing, and resumes from the committed offset after a simulated failure. Simulate a crash after processing the first batch, then restart and verify the consumer resumes from the correct offset rather than from zero, confirming at-least-once semantics are correctly implemented in the mock.
Analogy🏏Cricket
🏏 Think of it like cricket: the consumer is a scorer working through delivery slips over by over. An illegible slip does not stop play — it goes into the third umpire's query tray (the dead-letter queue) with a note of what was wrong, and scoring continues; one bad ball must never halt the match record. The offset commit discipline is the scorer's rule of only ruling off the book at the end of a completed over: mark your place only after the over is fully and correctly entered. So when the scorer is taken ill mid-session and a replacement steps in — the simulated crash — the newcomer opens the book at the last ruled-off over and resumes from there, not from ball one of the day. A couple of deliveries just before the interruption might get entered twice, which is exactly at-least-once delivery: the reconciliation may need a duplicate check, but no ball is ever lost. Commit-after-processing plus a query tray gives you a match record that survives any mid-innings disaster.
python
# exercise_kafka_pipeline.py — Step 2: Consumer with DLQ and crash recovery
class IPLEventConsumer:
DLQ_TOPIC = "ipl-deliveries-dlq"
def __init__(self, kafka: MockKafka, group_id: str):
self.kafka = kafka
self.group_id = group_id
self.processed = 0
self.dlq_count = 0
self.results: list[dict] = [] # simulated downstream write
def _validate_event(self, event: dict) -> bool:
"""Validate delivery event — returns False for unprocessable events."""
required = ["delivery_id","match_id","bowler","runs"]
if not all(k in event for k in required):
return False
if not (0 <= event.get("runs", -1) <= 6):
return False
return True
def process_batch(
self, topic: str, max_records: int = 20
) -> tuple[int, int]:
"""
Consume one batch: validate, write results, commit offsets.
Returns (processed_count, dlq_count).
At-least-once: commit AFTER write, not before.
"""
batch = self.kafka.consume_batch(topic, self.group_id, max_records)
if not batch:
return 0, 0
good, bad = [], []
for msg in batch:
if self._validate_event(msg.value):
good.append(msg)
else:
bad.append(msg)
# Route bad events to DLQ
for msg in bad:
self.kafka.produce(self.DLQ_TOPIC, msg.key,
{"original": msg.value, "error": "validation_failed"})
# Simulate downstream write (in production: write to DB/Delta)
for msg in good:
self.results.append(msg.value)
# Commit offsets AFTER downstream write — at-least-once guarantee
self.kafka.commit(self.group_id, batch)
self.processed += len(good)
self.dlq_count += len(bad)
return len(good), len(bad)
# Run consumer — first batch
consumer = IPLEventConsumer(mock_kafka, "ipl-analytics")
n_good, n_bad = consumer.process_batch("ipl-deliveries", max_records=20)
print(f"Batch 1: {n_good} processed, {n_bad} to DLQ")
# Inject a malformed event
mock_kafka.produce("ipl-deliveries", "10001:BAD",
{"delivery_id": 999, "runs": 99}) # runs=99 is invalid
# Simulate crash — consumer destroyed without committing current state
print("\nSimulating consumer crash...")
lag_before_crash = mock_kafka.lag("ipl-deliveries", "ipl-analytics")
print(f"Lag before crash: {lag_before_crash}")
# Restart consumer with same group_id — resumes from committed offset
restarted_consumer = IPLEventConsumer(mock_kafka, "ipl-analytics")
n_good2, n_bad2 = restarted_consumer.process_batch("ipl-deliveries", max_records=100)
print(f"After restart: {n_good2} processed, {n_bad2} to DLQ (includes malformed event)")
# Verify: bad event landed in DLQ
dlq_messages = mock_kafka.topics["ipl-deliveries-dlq"]
total_dlq = sum(len(msgs) for msgs in dlq_messages.values())
assert total_dlq >= 1, "Expected at least one DLQ message"
print(f"\nDLQ messages: {total_dlq}")
# Verify: total processed across both runs = all valid events
total_valid = producer.produced # injected 1 malformed, so valid = produced - 1
total_processed = consumer.processed + restarted_consumer.processed
print(f"Total produced: {producer.produced + 1} Total processed: {total_processed} DLQ: {total_dlq}")
assert total_processed + total_dlq == producer.produced + 1
print("\nAt-least-once accounting verified ✓")