In this post we are going to cover real-world manager scenarios 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: in the final round, the hiring manager asks you to design a real-time data platform on Azure for an online retailer.
These are the questions a manager asks when something already went wrong: lag at peak, duplicates in the warehouse, partitions moving around, a cost spike. Interviewers ask them because they show whether you have run Event Hubs in production, not just read about it. Each answer starts with how we measure the problem, then the fix.
Part 10 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: Consumer lag grows every evening at peak. How do you investigate and fix it?
- Question 2: The same events show up twice in the warehouse. Where do duplicates come from, and how do you stop them?
- Question 3: Partitions keep moving between consumer instances and processing stalls. How do you stop the rebalancing?
- Question 4: Event Hubs costs jumped this month. Where do you look first?
- Question 5: A device team asks whether to use IoT Hub or Event Hubs. How do you decide?
- Question 6: Events from a partner must reach your event hub safely. How do you set it up?
- Scenario 7: In the final round, the hiring manager asks you to design a real-time data platform on Azure for an online retailer. What is your design?
- The full script
The short answer
Measure lag per partition (last enqueued sequence number minus the checkpoint), expect duplicates from at-least-once delivery and make writes idempotent on a business key, keep processor instances stable, read cost by meter, and give partners send-only access to their own event hub.
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 Entra ID roles and Kafka consumer settings, are shown as code. Config.json creates four event hubs: orders with 4 partitions, sales with 2, partner-orders with 2 and checkout with 4. schema.sql creates two small tables, sales with a plain insert and sales_keyed with order_id as the primary key, and setup.py creates the checkpoint containers and the database.
{
"UserConfig": {
"NamespaceConfig": [
{
"Type": "EventHub",
"Name": "emulatorNs1",
"Entities": [
{
"Name": "orders",
"PartitionCount": 4,
"ConsumerGroups": [
{
"Name": "warehouse"
},
{
"Name": "rebal"
}
]
},
{
"Name": "sales",
"PartitionCount": 2,
"ConsumerGroups": [
{
"Name": "loader"
}
]
},
{
"Name": "partner-orders",
"PartitionCount": 2,
"ConsumerGroups": []
},
{
"Name": "checkout",
"PartitionCount": 4,
"ConsumerGroups": [
{
"Name": "fraud"
},
{
"Name": "analytics"
}
]
}
]
}
],
"LoggingConfig": {
"Type": "File"
}
}
}CREATE TABLE sales (order_id INTEGER); -- plain
CREATE TABLE sales_keyed (order_id INTEGER PRIMARY KEY); -- keyedimport os, sqlite3
from azure.storage.blob import BlobServiceClient
svc = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
for name in ("lag-checkpoints", "ckpt-plain", "ckpt-idempotent", "lb-ckpt"):
svc.create_container(name)
with sqlite3.connect("dw.db") as db:
db.executescript(open("schema.sql").read())Question 1: Consumer lag grows every evening at peak. How do you investigate and fix it?
First, measure lag per partition: the last enqueued sequence number minus the sequence number your consumer group last checkpointed. A single total hides where the problem is.
Let's create the problem. We pin 30 orders to partition 2 by partition ID, and send 10 to each of the others. The warehouse consumer, capped at 8 per partition for the demo, checkpoints. Then we check the lag. We run python send_orders.py, python slow_consumer.py and python lag.py.
import json, os
from azure.eventhub import EventData, EventHubProducerClient
load = {"0": 10, "1": 10, "2": 30, "3": 10} # partition 2 runs hot
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
order_id = 1000
with producer:
for pid, count in load.items():
batch = producer.create_batch(partition_id=pid)
for _ in range(count):
order_id += 1
batch.add(EventData(json.dumps({"order_id": order_id})))
producer.send_batch(batch)
print("sent per partition:", load)import logging, os, threading
from collections import Counter
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "lag-checkpoints")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="warehouse",
eventhub_name="orders", checkpoint_store=store)
done = Counter()
def on_event(pc, event):
if done[pc.partition_id] < 8: # keeps up with 8 per partition
done[pc.partition_id] += 1
if done[pc.partition_id] == 8:
pc.update_checkpoint(event) # one checkpoint per 8 events
threading.Timer(8, client.close).start()
with client:
client.receive(on_event, starting_position="-1")
print("processed and checkpointed:", dict(sorted(done.items())))import os
from azure.eventhub import EventHubProducerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "lag-checkpoints")
done = {c["partition_id"]: c["sequence_number"] for c in
store.list_checkpoints("localhost", "orders", "warehouse")}
client = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
with client:
for p in client.get_partition_ids():
props = client.get_partition_properties(p)
last, ckpt = props["last_enqueued_sequence_number"], done.get(p, -1)
print(f"partition {p}: last {last:2} checkpoint {ckpt:2}"
f" lag {last - ckpt}")$ python send_orders.py
sent per partition: {'0': 10, '1': 10, '2': 30, '3': 10}
$ python slow_consumer.py
processed and checkpointed: {'0': 8, '1': 8, '2': 8, '3': 8}
$ python lag.py
partition 0: last 9 checkpoint 7 lag 2
partition 1: last 9 checkpoint 7 lag 2
partition 2: last 29 checkpoint 7 lag 22
partition 3: last 9 checkpoint 7 lag 2We sent {'0': 10, '1': 10, '2': 30, '3': 10}, and the consumer checkpointed 8 on each. The lag is 2, 2, 22, 2. Partition 2 shows last 29 checkpoint 7 lag 22. One hot partition, not a slow consumer everywhere.
In Azure, Premium and Dedicated can log ConsumerLag in ApplicationMetricsLogs. For AMQP it is delivery lag, not checkpoint based. On any tier, the client can set track_last_enqueued_event_properties=True and report its own lag.
The fix depends on the cause:
| Cause | Fix |
|---|---|
| One hot partition | find the hot key or the pinned partition ID, and spread it |
| All partitions behind | more instances (up to one per partition), batch writes, checkpoint every N |
| Namespace at its limit | more TUs or PUs; egress slows with no error: watch OutgoingBytes |
Adding more instances than partitions. In one consumer group, the partition count is the ceiling for parallel consumers, and extra instances sit idle.
Question 2: The same events show up twice in the warehouse. Where do duplicates come from, and how do you stop them?
Event Hubs is at least once, so duplicates are normal. They come from three places:
- Consumer restarts - before its next checkpoint
- Partition changes owner - the new owner starts at the last checkpoint
- Producer retries - a retried send can publish the event twice
Let's see the first one, for real. The loader checkpoints every four events. It inserts six orders, checkpoints at sequence 3, then crashes. Then we restart it and count the rows. We run python send_sales.py, CRASH_AFTER=6 python loader.py plain, python loader.py plain and python report.py.
import os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="sales")
with producer:
batch = producer.create_batch(partition_id="0")
for order_id in range(5001, 5007):
batch.add(EventData(str(order_id)))
producer.send_batch(batch)
print("sent orders 5001 to 5006")import logging, os, sqlite3, sys, threading
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
mode, crash_after = sys.argv[1], int(os.environ.get("CRASH_AFTER", 0))
SQL = {"plain": "INSERT INTO sales VALUES (?)",
"idempotent": "INSERT INTO sales_keyed VALUES (?)"
" ON CONFLICT (order_id) DO NOTHING"}[mode]
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], f"ckpt-{mode}")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="loader",
eventhub_name="sales", checkpoint_store=store,
load_balancing_interval=2) # short for the demo; default 30 s
db = sqlite3.connect("dw.db", isolation_level=None) # autocommit
def on_event(pc, event):
order_id, seq = int(event.body_as_str()), event.sequence_number
added = db.execute(SQL, (order_id,)).rowcount
print(f"seq {seq} order {order_id}:", "inserted" if added else "skipped")
if seq % 4 == 3: # every 4 events
pc.update_checkpoint(event)
print(" checkpoint at seq", seq)
if seq + 1 == crash_after:
print(" crash before the next checkpoint"); os._exit(1)
threading.Timer(20, client.close).start() # a crashed owner expires
with client:
client.receive(on_event, starting_position="-1")import sqlite3
db = sqlite3.connect("dw.db")
for table in ("sales", "sales_keyed"):
rows, orders = db.execute("SELECT COUNT(*), COUNT(DISTINCT order_id)"
f" FROM {table}").fetchone()
print(f"{table}: {rows} rows, {orders} distinct orders")$ python send_sales.py
sent orders 5001 to 5006
$ CRASH_AFTER=6 python loader.py plain
seq 0 order 5001: inserted
seq 1 order 5002: inserted
seq 2 order 5003: inserted
seq 3 order 5004: inserted
checkpoint at seq 3
seq 4 order 5005: inserted
seq 5 order 5006: inserted
crash before the next checkpoint
$ python loader.py plain
seq 4 order 5005: inserted
seq 5 order 5006: inserted
$ python report.py
sales: 8 rows, 6 distinct orders
sales_keyed: 0 rows, 0 distinct ordersAfter checkpoint at seq 3 and crash before the next checkpoint, the restart resumes after the checkpoint and inserts order 5005 and order 5006 again. The result: sales: 8 rows, 6 distinct orders.
The fix is an idempotent write. order_id is the primary key, and the insert says on conflict do nothing. Same crash, same restart, into sales_keyed:
$ CRASH_AFTER=6 python loader.py idempotent
seq 0 order 5001: inserted
seq 1 order 5002: inserted
seq 2 order 5003: inserted
seq 3 order 5004: inserted
checkpoint at seq 3
seq 4 order 5005: inserted
seq 5 order 5006: inserted
crash before the next checkpoint
$ python loader.py idempotent
seq 4 order 5005: skipped
seq 5 order 5006: skipped
$ python report.py
sales: 8 rows, 6 distinct orders
sales_keyed: 6 rows, 6 distinct ordersThis time the two re-read orders are skipped, and the keyed table has sales_keyed: 6 rows, 6 distinct orders.
Key on a business value like order_id, not the sequence number. A producer retry creates a new event with a new sequence number, and only the business key catches it. In a warehouse, that is a MERGE on the business key.
The Python client has no idempotent producer, so the consumer side is where duplicates stop.
Question 3: Partitions keep moving between consumer instances and processing stalls. How do you stop the rebalancing?
Some rebalancing is normal. When instances start or stop, the processor moves partitions until each owns a fair share. If it lasts more than about five minutes, something is wrong.
Let's see it settle. balance.py runs two processor instances on four partitions, with a 2 second load balancing interval to keep the demo short. Python's default is 30 seconds. Then one instance stops cleanly. We run python balance.py.
import collections, logging, os, threading, time
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "lb-ckpt")
def counts(): # partitions per live instance
owners = collections.Counter(o["owner_id"] for o in
store.list_ownership("localhost", "orders", "rebal") if o["owner_id"])
return sorted(list(owners.values()) + [0] * (2 - len(owners)))
def start(): # one processor instance
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="rebal",
eventhub_name="orders", checkpoint_store=store,
load_balancing_interval=2, # default: 30 seconds
load_balancing_strategy="greedy") # the Python default
threading.Thread(target=client.receive, daemon=True,
kwargs=dict(on_event=lambda pc, e: None)).start()
return client
a, b = start(), start()
time.sleep(16)
print("two instances:", counts())
b.close() # clean shutdown releases
time.sleep(8)
print("after one stops:", counts())
a.close()$ python balance.py
two instances: [2, 2]
after one stops: [0, 4]The two instances settle at [2, 2]. When one shuts down cleanly, it releases its partitions, and the other takes all four: [0, 4].
Constant moves usually mean instances that keep crashing or restarting. A crash is worse than a clean stop: its partitions wait for the ownership to expire, 180 seconds by default in Python, which is 6 times the 30 second load balancing interval.
So fix the cause:
- Instances that don't die - enough memory and time per batch
- Close the client on shutdown - partitions are released at once
- A stable instance count - no scaling up and down every few minutes
- No more instances than partitions - extra ones sit idle
On the Kafka endpoint, check the session timeout and use static membership:
# Kafka endpoint consumers: settings that stop constant rebalancing
session.timeout.ms=30000
heartbeat.interval.ms=3000
max.poll.interval.ms=300000
request.timeout.ms=60000
# static membership: a fixed, unique id per instance
group.instance.id=loader-1Every move can redeliver events since the last checkpoint, one more reason writes must be idempotent.
Question 4: Event Hubs costs jumped this month. Where do you look first?
Start in cost analysis, grouped by meter, because each tier bills different things. We don't quote prices here: they change, the meters don't.
| Tier | Billed per hour | Also billed | Included |
|---|---|---|---|
| Basic, Standard | throughput units (TUs) | ingress events per million | 84 GB storage per TU |
| Premium | processing units (PUs) | storage above 1 TB per PU | ingress, Capture |
| Dedicated | capacity units (CUs), 4-hour minimum | storage above 10 TB per CU | ingress, Capture |
The usual surprise is auto-inflate. It raises throughput units at peak and never scales them back down by itself, so after one busy evening we keep paying for the higher count. Here is the full list of meters to check:
- Auto-inflate - raises TUs, never scales them down by itself
- Event size - 64 KB units: a 96 KB event bills as two ingress events
- Capture on Standard - billed per throughput unit
- Storage - beyond 84 GB per TU on Standard, from long retention
- Geo-replication - secondary compute + replication bandwidth
- Brokered connections - overage charge beyond the Standard quota
And a Dedicated cluster bills at least four hours.
Throughput units over the month, the ingress events trend, and any feature switched on recently. Then compare tiers for the load.
Question 5: A device team asks whether to use IoT Hub or Event Hubs. How do you decide?
Ask what the devices need besides sending data. Big streams from services and apps: Event Hubs. Real devices that need their own identity, commands from the cloud or management: IoT Hub.
| IoT Hub | Event Hubs | |
|---|---|---|
| Device-to-cloud, AMQP, HTTPS | yes | yes |
| MQTT | yes | no (AMQP, Kafka, HTTPS) |
| Per-device identity | yes | no: SAS policy or Entra role |
| File upload, Device Provisioning Service | yes | no |
| Cloud-to-device messages | yes (standard tier) | no |
| Device twins and management, IoT Edge | yes (standard tier) | no |
Both take device-to-cloud telemetry over AMQP and HTTPS. Only IoT Hub speaks MQTT, gives each device its own identity, and offers file upload and the Device Provisioning Service. IoT Hub's standard tier adds cloud-to-device messaging, device twins and management, and IoT Edge. Event Hubs has none of that: every producer is just a client with a policy or a role.
Microsoft's own guidance: "Even if the only use case is device-to-cloud data ingestion, we highly recommend using IoT Hub." Per-device identity means one stolen device can be blocked alone.
Often we use both:
- Devices - MQTT, one identity each
- IoT Hub - identity, twins, commands
- Event Hubs - a consumer group per data team
IoT Hub routes the telemetry to Event Hubs, and the data teams read it there with their own consumer groups.
Question 6: Events from a partner must reach your event hub safely. How do you set it up?
Give the partner the smallest door possible: their own event hub, send rights only, nothing else in the namespace. Validate their events there before they reach your internal hubs.
The flow looks like this:
- Acme, the partner - send only
partner-orders- their own event hub- Validate - schema and business rules
orders- internal event hubs
Best option: the partner's multitenant Entra app, consented into your tenant, gets Azure Event Hubs Data Sender on that one event hub. No shared secret to leak. If they can't use Entra ID, create a SAS policy on that event hub with only the Send right. Both are Azure commands, shown as code:
# Best: the partner's multitenant Entra app, consented in your tenant
az role assignment create --assignee "$ACME_APP_ID" \
--role "Azure Event Hubs Data Sender" --scope "$PARTNER_HUB_ID"
# Fallback: a SAS policy on that event hub with Send only
az eventhubs eventhub authorization-rule create \
--resource-group rg-shop --namespace-name tbshop-events \
--eventhub-name partner-orders --name acme-send --rights SendEven then, don't hand over the key. We sign a SAS token for that event hub with an expiry, and the partner gets a connection string with only the token. We run python make_token.py and python partner_send.py.
import base64, hashlib, hmac, urllib.parse
from datetime import datetime, timezone
def sas_token(uri, key_name, key, expiry):
sr = urllib.parse.quote_plus(uri)
to_sign = (sr + "\n" + str(expiry)).encode()
sig = hmac.new(key.encode(), to_sign, hashlib.sha256).digest()
sig = urllib.parse.quote_plus(base64.b64encode(sig))
return (f"SharedAccessSignature sr={sr}&sig={sig}"
f"&se={expiry}&skn={key_name}")
expiry = 1798761600 # demo: fixed. Real: now + a few hours
token = sas_token("sb://localhost/partner-orders", "acme-send",
"ACME_SEND_KEY", expiry) # the key stays with you
with open("partner.conn", "w") as f: # this goes to the partner
f.write(f"Endpoint=sb://localhost/;SharedAccessSignature={token};"
"EntityPath=partner-orders;UseDevelopmentEmulator=true")
print("token for: sb://localhost/partner-orders, policy acme-send")
print("expires:", datetime.fromtimestamp(expiry, timezone.utc))from azure.eventhub import EventData, EventHubProducerClient
conn = open("partner.conn").read() # token only, no key
print("connection string holds a key:", "SharedAccessKey=" in conn)
producer = EventHubProducerClient.from_connection_string(conn)
with producer:
batch = producer.create_batch()
for order in ("acme-7001", "acme-7002"):
batch.add(EventData(order))
producer.send_batch(batch)
print("partner sent", len(batch), "events to partner-orders")$ python make_token.py
token for: sb://localhost/partner-orders, policy acme-send
expires: 2027-01-01 00:00:00+00:00
$ python partner_send.py
connection string holds a key: False
partner sent 2 events to partner-ordersThe token is for sb://localhost/partner-orders, policy acme-send, and it expires: 2027-01-01 00:00:00+00:00. On the partner side, connection string holds a key: False, and it prints partner sent 2 events to partner-orders. The emulator doesn't check signatures, so this shows the client side only.
Two more things to plan:
- Not private? - IP firewall: default Deny + the partner's ranges
- Rotation - re-sign with the secondary key, then regenerate the primary
Regenerating a key instantly invalidates every token signed with it, so plan rotation before go-live.
Scenario 7: In the final round, the hiring manager asks you to design a real-time data platform on Azure for an online retailer. What is your design?
Start from the consumers. Fraud needs seconds, dashboards a minute, and the warehouse needs everything long term. So one namespace, event hubs for checkout, clicks and payments, and a consumer group per application.
Here is the design, end to end:
- Web, app, payments - producers, partition key = customer
- Event Hubs namespace -
checkout, clicks, payments fraudconsumer group - seconds;analyticsconsumer group - dashboards- Stream Analytics or Databricks - fraud rules, windows
- Capture to ADLS Gen2 - raw events: bronze
- Databricks or Fabric - silver and gold
Producers send JSON with the customer ID as the partition key, so each customer's events stay in order. Let's prove it with one checkout stream and two consumer groups. We run python retail_stream.py.
import json, logging, os, threading
from collections import defaultdict
from azure.eventhub import (EventData, EventHubConsumerClient,
EventHubProducerClient)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name="checkout")
with producer:
for i in range(8): # 4 customers, 2 orders each
order = {"customer": f"cust-{i % 4 + 1}", "order_id": 9001 + i}
producer.send_event(EventData(json.dumps(order)),
partition_key=order["customer"])
for group in ("fraud", "analytics"): # one group per app
got = defaultdict(list)
client = EventHubConsumerClient.from_connection_string(
conn, consumer_group=group, eventhub_name="checkout")
def on_event(pc, e, got=got):
order = e.body_as_json()
got[order["customer"]].append(order["order_id"])
threading.Timer(6, client.close).start()
with client:
client.receive(on_event, starting_position="-1")
in_order = all(ids == sorted(ids) for ids in got.values())
print(f"{group}: {sum(map(len, got.values()))} orders,"
f" in order per customer: {in_order}")
print(" cust-1:", got["cust-1"])$ python retail_stream.py
fraud: 8 orders, in order per customer: True
cust-1: [9001, 9005]
analytics: 8 orders, in order per customer: True
cust-1: [9001, 9005]Both groups read all eight orders: fraud: 8 orders, in order per customer: True and analytics: 8 orders, in order per customer: True. For example, cust-1: [9001, 9005] in both. Consumers write idempotently by order_id.
Around it:
- Partitions - sized from load tests, peak plus growth
- Microsoft Entra ID - managed identities, private endpoints
- Schema Registry - contracts between teams
- Alerts - lag,
ThrottledRequests,CaptureBacklog
Then the trade-off:
| Choose | When |
|---|---|
| Premium | geo-replication, up to 90 days retention, customer-managed keys |
| Standard | up to 7 days retention, auto-inflate TUs, Capture billed per TU |
Premium if they need geo-replication or 90 day replay, Standard with auto-inflate if not.
Plan replay up front, so a consumer bug becomes a reprocessing job, not data loss.
Key points to remember
- Lag per partition: last enqueued sequence number minus the checkpoint
- Duplicates are normal: idempotent writes on a business key
- Stop rebalancing: stable instances, clean shutdown, no more than partitions
- Cost: read it by meter; auto-inflate never scales down
- Partners: their own event hub, send-only, Entra ID or a short-lived token
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 #10 - Real-world manager scenarios
# 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_10_Code.py.
# Entities in Config.json: orders (4 partitions, groups warehouse, rebal); sales (2 partitions, groups loader); partner-orders (2 partitions); checkout (4 partitions, groups fraud, analytics)
# 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('schema.sql', """\
CREATE TABLE sales (order_id INTEGER); -- plain
CREATE TABLE sales_keyed (order_id INTEGER PRIMARY KEY); -- keyed
""")
write('setup.py', """\
import os, sqlite3
from azure.storage.blob import BlobServiceClient
svc = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
for name in ("lag-checkpoints", "ckpt-plain", "ckpt-idempotent", "lb-ckpt"):
svc.create_container(name)
with sqlite3.connect("dw.db") as db:
db.executescript(open("schema.sql").read())
""")
run('python setup.py')
# Step 1: A hot partition: lag per partition = last enqueued sequence number minus the checkpoint
write('send_orders.py', """\
import json, os
from azure.eventhub import EventData, EventHubProducerClient
load = {"0": 10, "1": 10, "2": 30, "3": 10} # partition 2 runs hot
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
order_id = 1000
with producer:
for pid, count in load.items():
batch = producer.create_batch(partition_id=pid)
for _ in range(count):
order_id += 1
batch.add(EventData(json.dumps({"order_id": order_id})))
producer.send_batch(batch)
print("sent per partition:", load)
""")
write('slow_consumer.py', """\
import logging, os, threading
from collections import Counter
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "lag-checkpoints")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="warehouse",
eventhub_name="orders", checkpoint_store=store)
done = Counter()
def on_event(pc, event):
if done[pc.partition_id] < 8: # keeps up with 8 per partition
done[pc.partition_id] += 1
if done[pc.partition_id] == 8:
pc.update_checkpoint(event) # one checkpoint per 8 events
threading.Timer(8, client.close).start()
with client:
client.receive(on_event, starting_position="-1")
print("processed and checkpointed:", dict(sorted(done.items())))
""")
write('lag.py', """\
import os
from azure.eventhub import EventHubProducerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "lag-checkpoints")
done = {c["partition_id"]: c["sequence_number"] for c in
store.list_checkpoints("localhost", "orders", "warehouse")}
client = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
with client:
for p in client.get_partition_ids():
props = client.get_partition_properties(p)
last, ckpt = props["last_enqueued_sequence_number"], done.get(p, -1)
print(f"partition {p}: last {last:2} checkpoint {ckpt:2}"
f" lag {last - ckpt}")
""")
run('python send_orders.py')
run('python slow_consumer.py')
run('python lag.py')
# Step 2: A consumer crashes before its next checkpoint: a plain insert writes duplicates
write('send_sales.py', """\
import os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="sales")
with producer:
batch = producer.create_batch(partition_id="0")
for order_id in range(5001, 5007):
batch.add(EventData(str(order_id)))
producer.send_batch(batch)
print("sent orders 5001 to 5006")
""")
write('loader.py', """\
import logging, os, sqlite3, sys, threading
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
mode, crash_after = sys.argv[1], int(os.environ.get("CRASH_AFTER", 0))
SQL = {"plain": "INSERT INTO sales VALUES (?)",
"idempotent": "INSERT INTO sales_keyed VALUES (?)"
" ON CONFLICT (order_id) DO NOTHING"}[mode]
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], f"ckpt-{mode}")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="loader",
eventhub_name="sales", checkpoint_store=store,
load_balancing_interval=2) # short for the demo; default 30 s
db = sqlite3.connect("dw.db", isolation_level=None) # autocommit
def on_event(pc, event):
order_id, seq = int(event.body_as_str()), event.sequence_number
added = db.execute(SQL, (order_id,)).rowcount
print(f"seq {seq} order {order_id}:", "inserted" if added else "skipped")
if seq % 4 == 3: # every 4 events
pc.update_checkpoint(event)
print(" checkpoint at seq", seq)
if seq + 1 == crash_after:
print(" crash before the next checkpoint"); os._exit(1)
threading.Timer(20, client.close).start() # a crashed owner expires
with client:
client.receive(on_event, starting_position="-1")
""")
write('report.py', """\
import sqlite3
db = sqlite3.connect("dw.db")
for table in ("sales", "sales_keyed"):
rows, orders = db.execute("SELECT COUNT(*), COUNT(DISTINCT order_id)"
f" FROM {table}").fetchone()
print(f"{table}: {rows} rows, {orders} distinct orders")
""")
run('python send_sales.py', allow_error=True)
run('CRASH_AFTER=6 python loader.py plain', allow_error=True)
run('python loader.py plain', allow_error=True)
run('python report.py', allow_error=True)
# Step 3: The same crash with an idempotent write keyed by order_id
run('CRASH_AFTER=6 python loader.py idempotent', allow_error=True)
run('python loader.py idempotent', allow_error=True)
run('python report.py', allow_error=True)
# Step 4: Two processor instances share four partitions; one stops cleanly
write('balance.py', """\
import collections, logging, os, threading, time
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "lb-ckpt")
def counts(): # partitions per live instance
owners = collections.Counter(o["owner_id"] for o in
store.list_ownership("localhost", "orders", "rebal") if o["owner_id"])
return sorted(list(owners.values()) + [0] * (2 - len(owners)))
def start(): # one processor instance
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="rebal",
eventhub_name="orders", checkpoint_store=store,
load_balancing_interval=2, # default: 30 seconds
load_balancing_strategy="greedy") # the Python default
threading.Thread(target=client.receive, daemon=True,
kwargs=dict(on_event=lambda pc, e: None)).start()
return client
a, b = start(), start()
time.sleep(16)
print("two instances:", counts())
b.close() # clean shutdown releases
time.sleep(8)
print("after one stops:", counts())
a.close()
""")
run('python balance.py')
# Step 5: You sign a SAS token for one event hub; the partner sends with the token only
write('make_token.py', 'import base64, hashlib, hmac, urllib.parse\nfrom datetime import datetime, timezone\n\ndef sas_token(uri, key_name, key, expiry):\n sr = urllib.parse.quote_plus(uri)\n to_sign = (sr + "\\n" + str(expiry)).encode()\n sig = hmac.new(key.encode(), to_sign, hashlib.sha256).digest()\n sig = urllib.parse.quote_plus(base64.b64encode(sig))\n return (f"SharedAccessSignature sr={sr}&sig={sig}"\n f"&se={expiry}&skn={key_name}")\n\nexpiry = 1798761600 # demo: fixed. Real: now + a few hours\ntoken = sas_token("sb://localhost/partner-orders", "acme-send",\n "ACME_SEND_KEY", expiry) # the key stays with you\nwith open("partner.conn", "w") as f: # this goes to the partner\n f.write(f"Endpoint=sb://localhost/;SharedAccessSignature={token};"\n "EntityPath=partner-orders;UseDevelopmentEmulator=true")\nprint("token for: sb://localhost/partner-orders, policy acme-send")\nprint("expires:", datetime.fromtimestamp(expiry, timezone.utc))\n')
write('partner_send.py', """\
from azure.eventhub import EventData, EventHubProducerClient
conn = open("partner.conn").read() # token only, no key
print("connection string holds a key:", "SharedAccessKey=" in conn)
producer = EventHubProducerClient.from_connection_string(conn)
with producer:
batch = producer.create_batch()
for order in ("acme-7001", "acme-7002"):
batch.add(EventData(order))
producer.send_batch(batch)
print("partner sent", len(batch), "events to partner-orders")
""")
run('python make_token.py')
run('python partner_send.py')
# Step 6: One checkout stream, keyed by customer, read by two consumer groups
write('retail_stream.py', """\
import json, logging, os, threading
from collections import defaultdict
from azure.eventhub import (EventData, EventHubConsumerClient,
EventHubProducerClient)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name="checkout")
with producer:
for i in range(8): # 4 customers, 2 orders each
order = {"customer": f"cust-{i % 4 + 1}", "order_id": 9001 + i}
producer.send_event(EventData(json.dumps(order)),
partition_key=order["customer"])
for group in ("fraud", "analytics"): # one group per app
got = defaultdict(list)
client = EventHubConsumerClient.from_connection_string(
conn, consumer_group=group, eventhub_name="checkout")
def on_event(pc, e, got=got):
order = e.body_as_json()
got[order["customer"]].append(order["order_id"])
threading.Timer(6, client.close).start()
with client:
client.receive(on_event, starting_position="-1")
in_order = all(ids == sorted(ids) for ids in got.values())
print(f"{group}: {sum(map(len, got.values()))} orders,"
f" in order per customer: {in_order}")
print(" cust-1:", got["cust-1"])
""")
run('python retail_stream.py')
All 10 parts are listed on the Real Time Azure Event Hubs Interview Questions and Answers page on this blog.
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.