In this post we are going to cover retention, Capture and replay in Azure Event Hubs. These are 7 real Event Hubs interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: clickstream events must be replayable for a week and archived for a year at the lowest cost.
Retention is how long Event Hubs keeps an event. Capture copies the stream into the data lake, and replay means reading old events again, usually after a bug. Interviewers ask about this because every pipeline breaks at some point, and the real question is whether you can get the data back.
Part 6 of the Real Time Azure Event Hubs Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: Someone asks how long Event Hubs keeps events. How does retention work in each tier?
- Question 2: Raw events must be kept for a year in the data lake. How does Event Hubs Capture work?
- Question 3: A bug wrote wrong data for two hours. How do you replay events from Event Hubs?
- Question 4: A consumer was down for longer than the retention period. What happens, and how do you avoid it?
- Question 5: Your manager asks when to use log compaction. How does it work?
- Question 6: A new team needs all history, not just recent events. What are the options?
- Scenario 7: Clickstream events must be replayable for a week and archived for a year at the lowest cost. What is your design?
- The full script
The short answer
Retention is time based: 1 hour by default, up to 1 day on Basic, 7 days on Standard and 90 days on Premium and Dedicated, and reading never deletes. Capture archives the stream as Avro files to Blob Storage or ADLS Gen2. Replay = a new reader from a time, sequence number or offset plus idempotent writes; log compaction keeps the latest event per key.
Let's set up the project
We run Microsoft's official Event Hubs emulator locally with Azurite, and small Python scripts that use the azure-eventhub client, so every output is real. Features the emulator doesn't support, like Capture and retention settings, are shown as code. Config.json creates an orders event hub with 2 partitions and three consumer groups: warehouse, replay and newteam. setup.py creates the checkpoint container and a small sales table, send.py sends orders keyed by customer, and db.py writes to the table with an upsert keyed by order_id.
{
"UserConfig": {
"NamespaceConfig": [
{
"Type": "EventHub",
"Name": "emulatorNs1",
"Entities": [
{
"Name": "orders",
"PartitionCount": 2,
"ConsumerGroups": [
{
"Name": "warehouse"
},
{
"Name": "replay"
},
{
"Name": "newteam"
}
]
}
]
}
],
"LoggingConfig": {
"Type": "File"
}
}
}import os, sqlite3
from azure.storage.blob import BlobServiceClient
blobs = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
blobs.create_container("warehouse-checkpoints") # checkpoint store
with sqlite3.connect("warehouse.db") as db:
db.execute("CREATE TABLE sales"
" (order_id INTEGER PRIMARY KEY, amount INTEGER)")import json, os, sys
from azure.eventhub import EventData, EventHubProducerClient
first, count = int(sys.argv[1]), int(sys.argv[2])
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
with producer:
for order_id in range(first, first + count):
customer = ("alice", "bob")[order_id % 2]
amount = 10 + order_id % 10 * 5
order = {"order_id": order_id, "amount": amount}
producer.send_event(EventData(json.dumps(order)),
partition_key=customer)
print(f"sent orders {first} to {first + count - 1}")import sqlite3
def save(rows):
"""Upsert keyed by order_id: a replay is harmless."""
with sqlite3.connect("warehouse.db", timeout=10) as db:
db.executemany("INSERT INTO sales VALUES (?, ?)"
" ON CONFLICT (order_id) DO UPDATE"
" SET amount = excluded.amount", rows)
if __name__ == "__main__":
with sqlite3.connect("warehouse.db") as db:
rows = db.execute("SELECT * FROM sales ORDER BY 1").fetchall()
print("sales (order, amount):")
for i in range(0, len(rows), 4):
print(" ", rows[i:i + 4])Question 1: Someone asks how long Event Hubs keeps events. How does retention work in each tier?
Retention is time based. Each event stays for the event hub's retention period, then becomes unavailable, whether anyone read it or not.
The default is 1 hour. The maximum depends on the tier:
| Tier | Maximum retention | Retention storage included |
|---|---|---|
| Basic | 1 day | 84 GB per throughput unit |
| Standard | 7 days | 84 GB per throughput unit |
| Premium | 90 days | 1 TB per processing unit |
| Dedicated | 90 days | 10 TB per capacity unit |
Reading doesn't remove anything. Let's prove it. We send four orders with python send.py 1001 4, then run python peek.py, which reads the event hub twice from the start and prints each partition's first and last sequence number.
import logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
def client():
return EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="orders")
for attempt in (1, 2):
reader, seen = client(), []
threading.Timer(5, reader.close).start()
with reader:
reader.receive(on_event=lambda ctx, ev: seen.append(ev),
starting_position="-1") # from the start
print(f"read {attempt}: {len(seen)} events")
with client() as c:
for pid in c.get_partition_ids():
p = c.get_partition_properties(pid)
print(f"partition {pid}: first seq"
f" {p['beginning_sequence_number']},"
f" last seq {p['last_enqueued_sequence_number']}")$ python send.py 1001 4
sent orders 1001 to 1004
$ python peek.py
read 1: 4 events
read 2: 4 events
partition 0: first seq 0, last seq 1
partition 1: first seq 0, last seq 1We get read 1: 4 events and read 2: 4 events. Both partitions still show first seq 0, last seq 1. Reading consumed nothing.
Retention is set per event hub, in hours, with the Delete cleanup policy. Here it is as Bicep, set to the Standard maximum of 168 hours:
resource eh 'Microsoft.EventHub/namespaces/eventhubs@2026-01-01' = {
parent: ns
name: 'orders'
properties: {
partitionCount: 2
retentionDescription: {
cleanupPolicy: 'Delete'
retentionTimeInHours: 168 // 7 days: the Standard maximum
}
}
}A retention change applies to the events already stored, so shortening it affects data you already have.
The emulator's retention is fixed at 1 hour. In our tests, events were still readable after 56 minutes and gone by 61. That is real expiry, just too slow to show here.
Retention is for replay, not storage. Event Hubs isn't a permanent data store, so anything you need longer goes to Capture or your own lake.
Question 2: Raw events must be kept for a year in the data lake. How does Event Hubs Capture work?
Capture is built in. It writes the stream to Blob Storage or Data Lake Storage Gen2, with no code to run. It is on Standard, billed per throughput unit, and included on Premium and Dedicated. It is not on Basic.
It writes Avro files per partition, in windows: one by time and one by size. Whichever comes first closes the file.
| Window | Default | Range |
|---|---|---|
| Time | 5 minutes | 1 to 15 minutes |
| Size | 300 MB | 10 to 500 MB |
The emulator has no Capture, so here is the configuration as Bicep. It turns Capture on for clicks, with a 5 minute and 300 MB window, a managed identity for the storage account, and the file path.
// ns = the namespace, lake = the ADLS Gen2 storage account
var dir = '{Namespace}/{EventHub}/{PartitionId}'
var stamp = '{Year}/{Month}/{Day}/{Hour}/{Minute}/{Second}'
resource eh 'Microsoft.EventHub/namespaces/eventhubs@2026-01-01' = {
parent: ns
name: 'clicks'
properties: {
partitionCount: 8
captureDescription: {
enabled: true
encoding: 'Avro'
intervalInSeconds: 300 // time window: 5 minutes
sizeLimitInBytes: 314572800 // size window: 300 MB
skipEmptyArchives: true // no empty files
destination: {
name: 'EventHubArchive.AzureBlockBlob'
identity: { type: 'SystemAssigned' } // managed identity
properties: {
storageAccountResourceId: lake.id
blobContainer: 'capture'
archiveNameFormat: '${dir}/${stamp}'
}
}
}
}
}The path is namespace, event hub and partition, then year, month, day, hour, minute and second. You can use any order, but all of them are required. Each Avro record holds the sequence number, offset, enqueued time, properties, and the body as bytes:
record EventData (Microsoft.ServiceBus.Messaging)
SequenceNumber long
Offset string
EnqueuedTimeUtc string
SystemProperties map
Properties map
Body bytes (or null)Two details bite people. Capture writes empty files when there's no data, unless you set skipEmptyArchives. And it only captures events that arrive after you turn it on, so it is no backfill.
Give Capture a managed identity for the storage account, and let a lifecycle rule on the storage account handle old files.
Question 3: A bug wrote wrong data for two hours. How do you replay events from Event Hubs?
Fix the code first, then read the affected window again and overwrite the bad output. Within retention the events are still there, and a reader can start from an enqueued time, a sequence number, or an offset.
| Start from | Python starting_position | Use it when |
|---|---|---|
| Enqueued time | a datetime | you know when the bug went live |
| Sequence number | an int, per partition | you logged the last good event |
| Offset | a str, per partition | you stored offsets with your output |
| The beginning | "-1" | a full rebuild within retention |
Let's create the incident. First the correct warehouse job runs, then deploy.py logs the release time of version 2, four more orders arrive, and version 2 runs with a bug: amounts in cents. The commands are python warehouse.py, python deploy.py, python send.py 1005 4, BUG=1 python warehouse.py and python db.py.
import json, logging, os, threading
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
from db import save
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
BUG = os.environ.get("BUG") == "1" # v2: amounts in cents
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "warehouse-checkpoints")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="warehouse", eventhub_name="orders",
checkpoint_store=store)
done = []
def on_batch(context, events):
orders = [json.loads(e.body_as_str()) for e in events]
save([(o["order_id"], o["amount"] * (100 if BUG else 1))
for o in orders])
context.update_checkpoint(events[-1]) # once per batch
done.extend(o["order_id"] for o in orders)
threading.Timer(8, client.close).start()
with client:
client.receive_batch(on_event_batch=on_batch,
starting_position="-1")
version = "v2 with the bug" if BUG else "correct code"
print(f"warehouse ({version}): processed orders {sorted(done)}")import datetime, time
# stands in for the release pipeline: it logs when v2 went live
time.sleep(1)
now = datetime.datetime.now(datetime.timezone.utc)
open("release.txt", "w").write(now.isoformat())
time.sleep(1)
print("warehouse v2 deployed, release time logged")$ python warehouse.py
warehouse (correct code): processed orders [1001, 1002, 1003, 1004]
$ python deploy.py
warehouse v2 deployed, release time logged
$ python send.py 1005 4
sent orders 1005 to 1008
$ BUG=1 python warehouse.py
warehouse (v2 with the bug): processed orders [1005, 1006, 1007, 1008]
$ python db.py
sales (order, amount):
[(1001, 15), (1002, 20), (1003, 25), (1004, 30)]
[(1005, 3500), (1006, 4000), (1007, 4500), (1008, 5000)]The correct code processed [1001, 1002, 1003, 1004], with amounts 15 to 30. Version 2 processed [1005, 1006, 1007, 1008] and wrote 3500, 4000, 4500 and 5000: a hundred times too high.
Now the replay. replay.py starts at the logged release time with the fixed code. It uses its own consumer group, replay, with no checkpoint store, because a stored checkpoint wins over the starting position. We run python replay.py, then python db.py.
import datetime, json, logging, os, threading
from azure.eventhub import EventHubConsumerClient
from db import save
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
bad_from = datetime.datetime.fromisoformat(open("release.txt").read())
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="replay",
eventhub_name="orders") # no checkpoint store
fixed = []
def on_batch(context, events):
orders = [json.loads(e.body_as_str()) for e in events]
save([(o["order_id"], o["amount"]) for o in orders]) # fixed code
fixed.extend(o["order_id"] for o in orders)
threading.Timer(8, client.close).start()
with client:
client.receive_batch(on_event_batch=on_batch,
starting_position=bad_from)
print("replayed from the release time:", sorted(fixed))$ python replay.py
replayed from the release time: [1005, 1006, 1007, 1008]
$ python db.py
sales (order, amount):
[(1001, 15), (1002, 20), (1003, 25), (1004, 30)]
[(1005, 35), (1006, 40), (1007, 45), (1008, 50)]It prints replayed from the release time: [1005, 1006, 1007, 1008], and the table shows (1005, 35), (1006, 40), (1007, 45) and (1008, 50). Fixed.
That works because the write is an upsert keyed by order_id, so replaying a good event is harmless. One emulator note: a time start returns only later events, even with inclusive set.
On a real project, I do three things:
- Start a bit early - the bad code also processed any backlog waiting for it
- Log release times - so we know where to start
- Replay into keyed tables - or into a swap-in table
A stored checkpoint wins over the starting position, so a replay needs its own consumer group, or no checkpoint store.
Question 4: A consumer was down for longer than the retention period. What happens, and how do you avoid it?
The events it missed are gone. Once retention expires, no reader can get them back. A Kafka consumer whose committed offsets fell out of retention starts again wherever auto.offset.reset says.
So we catch it long before. Let's keep the warehouse down while four more orders arrive with python send.py 1009 4. Then python lag.py compares each partition's last sequence number with the warehouse checkpoint. Then we restart the warehouse and check the lag again.
import os
from azure.eventhub import (EventHubConsumerClient,
parse_connection_string)
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
namespace = parse_connection_string(conn).fully_qualified_namespace
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "warehouse-checkpoints")
done = {c["partition_id"]: c["sequence_number"] for c in
store.list_checkpoints(namespace, "orders", "warehouse")}
client = EventHubConsumerClient.from_connection_string(
conn, consumer_group="warehouse", eventhub_name="orders")
with client:
for pid in client.get_partition_ids():
props = client.get_partition_properties(pid)
last = props["last_enqueued_sequence_number"]
print(f"partition {pid}: last seq {last},"
f" checkpoint seq {done[pid]}, lag {last - done[pid]}")$ python send.py 1009 4
sent orders 1009 to 1012
$ python lag.py
partition 0: last seq 5, checkpoint seq 3, lag 2
partition 1: last seq 5, checkpoint seq 3, lag 2
$ python warehouse.py
warehouse (correct code): processed orders [1009, 1010, 1011, 1012]
$ python lag.py
partition 0: last seq 5, checkpoint seq 5, lag 0
partition 1: last seq 5, checkpoint seq 5, lag 0While the warehouse is down, both partitions show last seq 5, checkpoint seq 3, lag 2. When it restarts, it resumes from its checkpoint and processes only [1009, 1010, 1011, 1012], and the lag is back to lag 0. That is what a consumer sees after a normal restart: everything after its last checkpoint.
So alert on lag, and on consumer downtime, with thresholds well inside retention. On Premium and Dedicated, the application metrics logs also report ConsumerLag per partition and consumer group.
Keep retention long enough to cover a weekend outage, plus Capture, so a consumer that missed the window can reload from the lake.
Question 5: Your manager asks when to use log compaction. How does it work?
Use it when consumers need the latest value per key, not the full history: a customer profile, a price, a stock level. Event Hubs then keeps the last known value for each event key instead of deleting by time.
The key is the partition key on each event. To delete a key, we send a tombstone: that key with a null body. Let's look at five stock events for two products:
| Seq | Key | Before compaction |
|---|---|---|
| 0 | sku-1 | stock 10 |
| 1 | sku-2 | stock 4 |
| 2 | sku-1 | stock 7 |
| 3 | sku-2 | null (tombstone) |
| 4 | sku-1 | stock 5 |
After compaction, only the last value per key is left. The tombstone for sku-2 stays visible to consumers for the tombstone retention time:
| Seq | Key | After compaction |
|---|---|---|
| 3 | sku-2 | tombstone, until its retention ends |
| 4 | sku-1 | stock 5: the last value |
The compaction job runs on its own schedule, so between runs a key can still have several events. Sequence numbers and offsets never change, and a reader from the start sees at least the final state of every key.
We choose the Compact cleanup policy when we create the event hub, and it can't be changed later. Compaction settings don't run on the emulator, so here it is as Bicep:
resource eh 'Microsoft.EventHub/namespaces/eventhubs@2026-01-01' = {
parent: ns
name: 'inventory'
properties: {
partitionCount: 4
retentionDescription: {
cleanupPolicy: 'Compact' // only at creation
tombstoneRetentionTimeInHours: 24
}
}
}It is not on Basic, and compacted data is limited to 1 GB per partition on Standard and 250 GB per partition on Premium and Dedicated.
Kafka Connect's config, offset and status topics must use compaction. The mistake is compacting a stream you replay from, because compaction removes history.
Question 6: A new team needs all history, not just recent events. What are the options?
It depends on how far back they need to go:
- Within retention - a new consumer group, starting from the beginning
- Beyond retention - Capture files in the lake
- Longer retention - up to 90 days on Premium and Dedicated
Event Hubs is not the place to keep years.
The clean pattern is backfill, then hand off. Load the archive from Capture, note the last sequence number per partition, and start the stream right after it. newteam.py does both in its own newteam group. First it reads from the start of retention. Then it starts the stream after sequence 3 on both partitions, which stands in for where our Capture archive ended. We run python newteam.py.
import json, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
def read(starting_position):
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="newteam", eventhub_name="orders")
got = []
def on_event(ctx, event):
order = json.loads(event.body_as_str())["order_id"]
got.append((ctx.partition_id, event.sequence_number, order))
threading.Timer(6, client.close).start()
with client:
client.receive(on_event=on_event,
starting_position=starting_position)
return sorted(got)
print("from the start of retention:", len(read("-1")), "events")
archived = {"0": 3, "1": 3} # last seq loaded from Capture
for pid, seq, order in read(archived):
print(f"stream: partition {pid} seq {seq} order {order}")$ python newteam.py
from the start of retention: 12 events
stream: partition 0 seq 4 order 1010
stream: partition 0 seq 5 order 1012
stream: partition 1 seq 4 order 1009
stream: partition 1 seq 5 order 1011From the start of retention, the new team gets 12 events, all twelve orders, while the warehouse checkpoints stay where they are. The stream then starts at seq 4 on both partitions, with orders 1010 and 1012 on partition 0 and 1009 and 1011 on partition 1. No gap and no overlap.
That works because sequence numbers are per partition, and every Capture record carries one.
Pace the first full read: throughput units are shared by everyone on the namespace.
Scenario 7: Clickstream events must be replayable for a week and archived for a year at the lowest cost. What is your design?
Split the two needs.
A week of replay is the event hub's job. We use Standard, with retention of 168 hours, the Standard maximum. Premium only makes sense if we need its isolation or longer retention.
A year of archive is Capture's job: Avro files in Data Lake Storage, with a larger time window, like 15 minutes, for fewer, bigger files, and skipEmptyArchives on.
Here is the whole flow:
- Clickstream - web and app producers
clicksevent hub - Standard, 168 hours- Capture - Avro, 15 minute window
- ADLS Gen2 - one year of files
The cheap part of the year comes from the storage account. A lifecycle rule moves Capture files to cool after 30 days, to cold after 90, and deletes them after 365. This is storage configuration, shown as code:
{"rules": [{
"name": "clicks-capture-one-year",
"enabled": true, "type": "Lifecycle",
"definition": {
"filters": {"blobTypes": ["blockBlob"],
"prefixMatch": ["capture/tbshop-ns/clicks/"]},
"actions": {"baseBlob": {
"tierToCool": {"daysAfterModificationGreaterThan": 30},
"tierToCold": {"daysAfterModificationGreaterThan": 90},
"delete": {"daysAfterModificationGreaterThan": 365}}}}}]}On Standard, Capture is billed per throughput unit, so size the units for the real load and alert on the CaptureBacklog metric. Replay within a week reads Event Hubs. Older data is a batch job over the Avro files.
And say why not a year in Event Hubs: retention tops out at 90 days, and it isn't a permanent store.
Capture plus a lifecycle rule gives a week hot and a year cheap.
Key points to remember
- Retention is time based: 1 day Basic, 7 days Standard, 90 days Premium and Dedicated
- Capture: Avro files per partition, time or size window, to Blob Storage or ADLS Gen2
- Replay from a time, sequence number or offset into idempotent writes
- Alert on lag inside retention; Capture is the fallback for missed events
- Log compaction keeps the last value per key; set it at creation
The full script
This is a Python script that writes every file and runs every command above in order against the Event Hubs emulator: start the emulator with Azurite, run pip install azure-eventhub azure-eventhub-checkpointstoreblob azure-storage-blob confluent-kafka fastavro first, and run it in an empty folder.
Show the full script that writes the files and runs every step
# ========================================================================
# Event Hubs Interview Q&A #6 - Retention, Capture and replay
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Azure Event Hubs emulator (Docker image, version 2.2.1) with Azurite
# and azure-eventhub 5.15.1; the outputs in the video and the blog post are the real outputs.
# To run it yourself:
# 1. Start the emulator with Microsoft's installer (github.com/Azure/azure-event-hubs-emulator-installer),
# using a Config.json with the event hubs and consumer groups listed below.
# 2. pip install azure-eventhub azure-eventhub-checkpointstoreblob azure-storage-blob confluent-kafka fastavro
# 3. Set EVENTHUB_CONNECTION_STRING to the emulator connection string (Endpoint=sb://localhost;
# SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;)
# and STORAGE_CONNECTION_STRING to the Azurite blob connection string, then run python EventHubs_Interview_QA_06_Code.py.
# Entities in Config.json: orders (2 partitions, groups warehouse, replay, newteam)
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================
import os, subprocess, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: the starting files
write('setup.py', """\
import os, sqlite3
from azure.storage.blob import BlobServiceClient
blobs = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
blobs.create_container("warehouse-checkpoints") # checkpoint store
with sqlite3.connect("warehouse.db") as db:
db.execute("CREATE TABLE sales"
" (order_id INTEGER PRIMARY KEY, amount INTEGER)")
""")
write('send.py', """\
import json, os, sys
from azure.eventhub import EventData, EventHubProducerClient
first, count = int(sys.argv[1]), int(sys.argv[2])
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
with producer:
for order_id in range(first, first + count):
customer = ("alice", "bob")[order_id % 2]
amount = 10 + order_id % 10 * 5
order = {"order_id": order_id, "amount": amount}
producer.send_event(EventData(json.dumps(order)),
partition_key=customer)
print(f"sent orders {first} to {first + count - 1}")
""")
write('db.py', 'import sqlite3\n\ndef save(rows):\n """Upsert keyed by order_id: a replay is harmless."""\n with sqlite3.connect("warehouse.db", timeout=10) as db:\n db.executemany("INSERT INTO sales VALUES (?, ?)"\n " ON CONFLICT (order_id) DO UPDATE"\n " SET amount = excluded.amount", rows)\n\nif __name__ == "__main__":\n with sqlite3.connect("warehouse.db") as db:\n rows = db.execute("SELECT * FROM sales ORDER BY 1").fetchall()\n print("sales (order, amount):")\n for i in range(0, len(rows), 4):\n print(" ", rows[i:i + 4])\n')
run('python setup.py')
# Step 1: Send four orders, read them twice from the start: reading removes nothing
write('peek.py', """\
import logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
def client():
return EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="orders")
for attempt in (1, 2):
reader, seen = client(), []
threading.Timer(5, reader.close).start()
with reader:
reader.receive(on_event=lambda ctx, ev: seen.append(ev),
starting_position="-1") # from the start
print(f"read {attempt}: {len(seen)} events")
with client() as c:
for pid in c.get_partition_ids():
p = c.get_partition_properties(pid)
print(f"partition {pid}: first seq"
f" {p['beginning_sequence_number']},"
f" last seq {p['last_enqueued_sequence_number']}")
""")
run('python send.py 1001 4')
run('python peek.py')
# Step 2: The warehouse job: v1 is correct, v2 writes amounts in cents
write('warehouse.py', """\
import json, logging, os, threading
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
from db import save
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
BUG = os.environ.get("BUG") == "1" # v2: amounts in cents
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "warehouse-checkpoints")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="warehouse", eventhub_name="orders",
checkpoint_store=store)
done = []
def on_batch(context, events):
orders = [json.loads(e.body_as_str()) for e in events]
save([(o["order_id"], o["amount"] * (100 if BUG else 1))
for o in orders])
context.update_checkpoint(events[-1]) # once per batch
done.extend(o["order_id"] for o in orders)
threading.Timer(8, client.close).start()
with client:
client.receive_batch(on_event_batch=on_batch,
starting_position="-1")
version = "v2 with the bug" if BUG else "correct code"
print(f"warehouse ({version}): processed orders {sorted(done)}")
""")
write('deploy.py', """\
import datetime, time
# stands in for the release pipeline: it logs when v2 went live
time.sleep(1)
now = datetime.datetime.now(datetime.timezone.utc)
open("release.txt", "w").write(now.isoformat())
time.sleep(1)
print("warehouse v2 deployed, release time logged")
""")
run('python warehouse.py')
run('python deploy.py')
run('python send.py 1005 4')
run('BUG=1 python warehouse.py')
run('python db.py')
# Step 3: Replay from the release time with the fixed code: the upsert repairs the table
write('replay.py', """\
import datetime, json, logging, os, threading
from azure.eventhub import EventHubConsumerClient
from db import save
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
bad_from = datetime.datetime.fromisoformat(open("release.txt").read())
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="replay",
eventhub_name="orders") # no checkpoint store
fixed = []
def on_batch(context, events):
orders = [json.loads(e.body_as_str()) for e in events]
save([(o["order_id"], o["amount"]) for o in orders]) # fixed code
fixed.extend(o["order_id"] for o in orders)
threading.Timer(8, client.close).start()
with client:
client.receive_batch(on_event_batch=on_batch,
starting_position=bad_from)
print("replayed from the release time:", sorted(fixed))
""")
run('python replay.py')
run('python db.py')
# Step 4: The warehouse is down: lag per partition, then a restart resumes from the checkpoint
write('lag.py', """\
import os
from azure.eventhub import (EventHubConsumerClient,
parse_connection_string)
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
namespace = parse_connection_string(conn).fully_qualified_namespace
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "warehouse-checkpoints")
done = {c["partition_id"]: c["sequence_number"] for c in
store.list_checkpoints(namespace, "orders", "warehouse")}
client = EventHubConsumerClient.from_connection_string(
conn, consumer_group="warehouse", eventhub_name="orders")
with client:
for pid in client.get_partition_ids():
props = client.get_partition_properties(pid)
last = props["last_enqueued_sequence_number"]
print(f"partition {pid}: last seq {last},"
f" checkpoint seq {done[pid]}, lag {last - done[pid]}")
""")
run('python send.py 1009 4')
run('python lag.py')
run('python warehouse.py')
run('python lag.py')
# Step 5: A new team reads all retained history, then hands off from the archive by sequence number
write('newteam.py', """\
import json, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
def read(starting_position):
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="newteam", eventhub_name="orders")
got = []
def on_event(ctx, event):
order = json.loads(event.body_as_str())["order_id"]
got.append((ctx.partition_id, event.sequence_number, order))
threading.Timer(6, client.close).start()
with client:
client.receive(on_event=on_event,
starting_position=starting_position)
return sorted(got)
print("from the start of retention:", len(read("-1")), "events")
archived = {"0": 3, "1": 3} # last seq loaded from Capture
for pid, seq, order in read(archived):
print(f"stream: partition {pid} seq {seq} order {order}")
""")
run('python newteam.py')
Part 7: Schema Registry and Event Formats
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on the official Azure Event Hubs emulator (2.2.1) with Azurite and azure-eventhub 5.15.1. Features the emulator does not support are shown as code. Part of the Real Time Azure Event Hubs Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Azure Event Hubs documentation (learn.microsoft.com/azure/event-hubs) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.