In this post we are going to cover security, networking and disaster recovery 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: a bank needs a secure, highly available event platform across two regions.
Security here means three things: who can send and read, from which network, and with which keys the data is encrypted. Disaster recovery means what happens when a zone or a whole region goes down. Interviewers ask about this because these settings are hard to change later, and a wrong answer in production means a leaked key or lost events.
Part 9 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: Leadership asks how producers and consumers should authenticate. SAS keys or Microsoft Entra ID?
- Question 2: A pipeline needs access with no secrets in code. How do managed identities and Event Hubs roles work?
- Question 3: Security wants no public access to the namespace. What networking options do you set up?
- Question 4: A teammate asks what happens if an availability zone fails. How do zones protect Event Hubs?
- Question 5: The team needs a plan for a full region outage. What is the difference between Geo-disaster recovery and Geo-replication?
- Question 6: Someone asks how to encrypt data with the company's own keys. How does customer-managed key encryption work?
- Scenario 7: A bank needs a secure, highly available event platform across two regions. What do you set up?
- The full script
The short answer
Use Microsoft Entra ID with managed identities and the Azure Event Hubs Data Sender / Receiver roles, and disable SAS (local auth). Close the namespace with private endpoints and public access disabled. Availability zones are built in; Geo-DR copies metadata only, Geo-replication (Premium, Dedicated) copies events and offsets too.
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. Most of this topic, like Entra ID, private endpoints, customer-managed keys and geo-replication, doesn't run on the emulator, so it is shown as code. Config.json creates two event hubs with 4 partitions each: orders, with the consumer group fraud, and payments, with ops. setup.py creates the dr-checkpoints container in Azurite.
{
"UserConfig": {
"NamespaceConfig": [
{
"Type": "EventHub",
"Name": "emulatorNs1",
"Entities": [
{
"Name": "orders",
"PartitionCount": 4,
"ConsumerGroups": [
{
"Name": "fraud"
}
]
},
{
"Name": "payments",
"PartitionCount": 4,
"ConsumerGroups": [
{
"Name": "ops"
}
]
}
]
}
],
"LoggingConfig": {
"Type": "File"
}
}
}import os
from azure.storage.blob import BlobServiceClient
svc = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
svc.create_container("dr-checkpoints")Question 1: Leadership asks how producers and consumers should authenticate. SAS keys or Microsoft Entra ID?
Microsoft Entra ID, wherever the client supports it. Microsoft recommends it, and with a managed identity there's no secret to leak. SAS stays for the few clients that can't use it.
| SAS key or token | Microsoft Entra ID | |
|---|---|---|
| Secret to store | yes: a key or a token | no: a managed identity |
| Rights | Send, Listen, Manage per policy | Data Sender, Data Receiver, Data Owner roles |
| Revoke | regenerate the key: every token breaks | remove one role assignment |
| Who did it? | every app shares the policy | one identity per app |
Let's look at what a SAS connection string holds, and the fields of a SAS token. We run python sas_token.py.
import base64, hashlib, hmac, os, urllib.parse
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
parts = dict(p.split("=", 1) for p in conn.strip(";").split(";"))
for name, value in parts.items():
shown = "*****" if name == "SharedAccessKey" else value
print(f"{name:24} {shown}")
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}")
token = sas_token("sb://localhost/orders", "orders-send", "SEND_KEY",
1798761600) # expiry: 2027-01-01 UTC
for field in token.split(" ", 1)[1].split("&"):
print("token", field)$ python sas_token.py
Endpoint sb://localhost
SharedAccessKeyName RootManageSharedAccessKey
SharedAccessKey *****
UseDevelopmentEmulator true
token sr=sb%3A%2F%2Flocalhost%2Forders
token sig=cw%2FF6ab%2BhsDJyzHaiLDISr9UzQjOnIegqAjIGfq78N8%3D
token se=1798761600
token skn=orders-sendThe connection string has the Endpoint, the policy name, here RootManageSharedAccessKey, the key itself, which we mask as *, and the emulator flag UseDevelopmentEmulator true. Root manage can manage the whole namespace. Never put it in an application.
The token is signed with a policy key. It has four fields: sr is the resource, sig is the signature, se=1798761600 is the expiry in seconds since 1970, and skn=orders-send is the key name. Regenerating the key breaks every token signed with it.
One honest note: the emulator doesn't check keys, so we can't show a send-only policy being refused. In Azure, each policy has Send, Listen or Manage rights, with up to 12 policies per namespace or event hub.
One shared key for every app. Give each app its own identity, Kafka clients too through OAUTHBEARER, then set disableLocalAuth: true, so SAS stops working.
Question 2: A pipeline needs access with no secrets in code. How do managed identities and Event Hubs roles work?
A managed identity is an identity Azure gives the compute itself, like a function app or a virtual machine. The code asks Entra ID for a token at run time, so there's nothing to store.
Here is the flow:
orders-func- a managed identity, no secret- Microsoft Entra ID - issues a token at run time
ordersevent hub - role: Data Sender
In code, only the credential changes: the fully qualified namespace plus ManagedIdentityCredential, instead of a connection string. Locally, DefaultAzureCredential uses your login. The emulator has no Entra ID, so this is code only:
from azure.eventhub import EventData, EventHubProducerClient
from azure.identity import ManagedIdentityCredential
producer = EventHubProducerClient(
fully_qualified_namespace="tbshop-events.servicebus.windows.net",
eventhub_name="orders",
credential=ManagedIdentityCredential()) # no key, no secret
with producer:
batch = producer.create_batch()
batch.add(EventData('{"order_id": 1001}'))
producer.send_batch(batch)Then we grant a data role:
| Built-in role | Allows | Give it to |
|---|---|---|
| Azure Event Hubs Data Sender | send | producers |
| Azure Event Hubs Data Receiver | receive | consumers, Stream Analytics jobs |
| Azure Event Hubs Data Owner | complete access | admins and automation only |
Here the function's identity gets Data Sender on the orders event hub only. Use the narrowest scope that works, and allow up to 5 minutes for it to take effect.
PRINCIPAL_ID=$(az functionapp identity assign \
--name orders-func --resource-group rg-shop \
--query principalId --output tsv)
HUB_ID=$(az eventhubs eventhub show --resource-group rg-shop \
--namespace-name tbshop-events --name orders \
--query id --output tsv)
az role assignment create \
--assignee-object-id "$PRINCIPAL_ID" \
--assignee-principal-type ServicePrincipal \
--role "Azure Event Hubs Data Sender" --scope "$HUB_ID"The consumer has Data Receiver and still fails, because ownership and checkpoints live in blob storage. Give the same identity a blob data role there too, like Storage Blob Data Contributor on the checkpoint container.
Question 3: Security wants no public access to the namespace. What networking options do you set up?
By default, a namespace is reachable from the internet, as long as the caller authenticates. So we add a private endpoint and disable public network access. Both need Standard tier or higher, so not Basic.
Here is the target picture:
- Shop apps - in
vnet-data, subnetsnet-apps - Private endpoint - a private IP, group ID
namespace tbshop-events- public network accessDisabled- Internet - no route in
The private endpoint gives the namespace a private IP in your virtual network. DNS goes through the zone privatelink.servicebus.windows.net, so the normal host name resolves privately.
az network private-endpoint create --name pe-tbshop-events \
--resource-group rg-shop --vnet-name vnet-data --subnet snet-pe \
--private-connection-resource-id "$NAMESPACE_ID" \
--group-id namespace --connection-name tbshop-events
az network private-dns zone create --resource-group rg-shop \
--name privatelink.servicebus.windows.net
# then link the zone to the VNet and add a DNS zone groupIn Bicep: public network access disabled, default action deny, and local auth off while we're at it. If some public access must stay, list only your ranges in the IP firewall.
resource ns 'Microsoft.EventHub/namespaces@2026-01-01' = {
name: 'tbshop-events'
location: location
sku: { name: 'Premium', tier: 'Premium', capacity: 1 }
properties: {
publicNetworkAccess: 'Disabled'
disableLocalAuth: true
}
resource rules 'networkRuleSets' = {
name: 'default'
properties: {
publicNetworkAccess: 'Disabled'
defaultAction: 'Deny'
trustedServiceAccessEnabled: true
ipRules: [] // allowed ranges, only if public stays
}
}
}Trusted Microsoft services in the same tenant can bypass the firewall if you allow it:
- Stream Analytics - with managed identity
- Event Grid, IoT Hub
- Azure Monitor - diagnostic settings
Azure Functions isn't one of them, so it needs VNet integration.
DNS still resolves to the public address, and every connection fails once public access is off. Test name resolution from inside the VNet first.
Question 4: A teammate asks what happens if an availability zone fails. How do zones protect Event Hubs?
Zone redundancy is built in. In regions with zones, every tier spreads the namespace across zones at no extra cost, and each event is replicated across zones before the send is acknowledged. So no acknowledged event is lost.
Clients see a short blip: a few seconds of downtime, and active requests can drop. So producers must retry. Python retries 3 times by default, backing off from 0.8 seconds. Our producer allows 5.
Now let's look at how we send. spread.py sends eight payments with no partition information, then eight more pinned to partition 2, and prints the event count per partition. We run python spread.py.
import os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="payments",
retry_total=5, retry_backoff_factor=0.8) # ride out a zone blip
def counts(): # events per partition
return [producer.get_partition_properties(p)
["last_enqueued_sequence_number"] + 1
for p in producer.get_partition_ids()]
with producer:
for i in range(8): # no key, no partition id
producer.send_event(EventData(f"payment-{i}"))
print("no partition info:", counts())
for i in range(8, 16): # pinned to one partition
producer.send_event(EventData(f"payment-{i}"), partition_id="2")
print("partition_id 2: ", counts())$ python spread.py
no partition info: [2, 2, 2, 2]
partition_id 2: [2, 2, 10, 2]With no partition info, the service spreads them: [2, 2, 2, 2] on this run. Pinned to partition 2, all eight land there: [2, 2, 10, 2].
That choice decides availability. If a partition is unavailable, events without partition information go to the remaining ones. Pinned to a partition ID or key, you get an error instead. We can't fail a zone on a laptop, so that part is from the docs.
Send without partition information unless ordering needs a key. And on Dedicated, zones need at least 3 capacity units.
Question 5: The team needs a plan for a full region outage. What is the difference between Geo-disaster recovery and Geo-replication?
Geo-disaster recovery copies metadata only: the event hubs, consumer groups and settings, never the events. Geo-replication copies everything: metadata, events and consumer offsets.
| Geo-disaster recovery | Geo-replication | |
|---|---|---|
| Copies | metadata only: event hubs, consumer groups, settings | metadata, events and consumer offsets |
| Tiers | Standard, Premium, Dedicated | Premium and Dedicated |
| Clients use | an alias: one stable host name | the namespace host name: the primary |
| Failover | once only, then the pairing is removed | promote: planned or forced, by you |
| Consumers after | old offsets are useless: reset them | offsets replicated; AMQP checkpoints in region 2 |
Geo-DR works from Standard up. Clients connect to an alias over two paired namespaces. Failover is once only: we run it on the secondary, the alias moves, and the pairing is removed. Role assignments aren't copied.
# pair: run on the primary, with the secondary as the partner
az eventhubs georecovery-alias set --resource-group rg-shop \
--namespace-name tbshop-events-east --alias tbshop-events \
--partner-namespace "$SECONDARY_NAMESPACE_ID"
# outage: fail over, run on the secondary (once only)
az eventhubs georecovery-alias fail-over --resource-group rg-shop \
--namespace-name tbshop-events-west --alias tbshop-eventsGeo-replication is generally available on Premium and Dedicated. Clients keep the namespace host name, which points to the primary. Replication is synchronous, with RPO 0 and more latency, or asynchronous, with a maximum lag. We promote a secondary ourselves, planned or forced.
Now the Geo-DR trap, on the emulator. First, the fraud consumer reads six orders and checkpoints. We run python send.py 1001 6 and python consume.py.
import 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:
batch = producer.create_batch(partition_id="0")
for order_id in range(first, first + count):
batch.add(EventData(f"order-{order_id}"))
producer.send_batch(batch)
print(f"sent order-{first} to order-{first + count - 1}")import logging, os, threading
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"], "dr-checkpoints")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="fraud",
eventhub_name="orders", checkpoint_store=store)
got = []
def on_event(pc, event):
got.append((pc.partition_id, event.sequence_number,
event.body_as_str()))
pc.update_checkpoint(event)
threading.Timer(8, client.close).start()
with client:
client.receive(on_event, starting_position="-1") # if no checkpoint
for p, seq, body in sorted(got):
print(f"read partition {p} seq {seq} {body}")
print("events read:", len(got))
for c in store.list_checkpoints("localhost", "orders", "fraud"):
print("checkpoint partition", c["partition_id"] + ":",
"seq", c["sequence_number"])$ python send.py 1001 6
sent order-1001 to order-1006
$ python consume.py
read partition 0 seq 0 order-1001
read partition 0 seq 1 order-1002
read partition 0 seq 2 order-1003
read partition 0 seq 3 order-1004
read partition 0 seq 4 order-1005
read partition 0 seq 5 order-1006
events read: 6
checkpoint partition 0: seq 5It reads order-1001 to order-1006 on partition 0, events read: 6, and saves checkpoint partition 0: seq 5.
Then we restart the emulator. A restart keeps the event hubs but drops the events, like a Geo-DR secondary after a failover. The checkpoints in Azurite stay. Three new orders arrive, and we run the consumer again. Then reset.py deletes the old checkpoint, and we run the consumer once more.
import os
from azure.storage.blob import ContainerClient
container = ContainerClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "dr-checkpoints")
prefix = "localhost/orders/fraud/checkpoint/"
for blob in container.list_blobs(name_starts_with=prefix):
container.delete_blob(blob.name)
print("deleted", blob.name)$ python send.py 2001 3
sent order-2001 to order-2003
$ python consume.py
events read: 0
checkpoint partition 0: seq 5
$ python reset.py
deleted localhost/orders/fraud/checkpoint/0
$ python consume.py
read partition 0 seq 0 order-2001
read partition 0 seq 1 order-2002
read partition 0 seq 2 order-2003
events read: 3
checkpoint partition 0: seq 2The first run prints events read: 0, silently, because the old checkpoint still says seq 5. After deleted localhost/orders/fraud/checkpoint/0, all three orders are read, order-2001 to order-2003.
Old offsets can't be reused. Reset the checkpoints and restart from the start, the end or a point in time.
Question 6: Someone asks how to encrypt data with the company's own keys. How does customer-managed key encryption work?
Data at rest is always encrypted with Microsoft-managed keys. A customer-managed key in Azure Key Vault adds a key you own on top, on Premium and Dedicated only.
There are three pieces:
tbshop-events- a Premium namespace- A managed identity - with access to the key: on an RBAC vault, the Key Vault Crypto Service Encryption User role
tbshop-kv- a key vault with soft delete and purge protection
The emulator has no encryption settings, so this is code. First, the vault, the key and the role for the identity:
az keyvault create --name tbshop-kv --resource-group rg-shop \
--location eastus2 --enable-purge-protection true \
--enable-rbac-authorization true
az keyvault key create --vault-name tbshop-kv --name eh-cmk --kty RSA
az role assignment create --assignee-object-id "$IDENTITY_PRINCIPAL_ID" \
--assignee-principal-type ServicePrincipal --scope "$KV_ID" \
--role "Key Vault Crypto Service Encryption User"Then the encryption block points at the vault URI, the key name and the identity. Do it at creation: it only works on a new or empty namespace, and fails if event hubs exist.
resource ns 'Microsoft.EventHub/namespaces@2026-01-01' = {
name: 'tbshop-events'
location: location
sku: { name: 'Premium', tier: 'Premium', capacity: 1 }
identity: {
type: 'UserAssigned'
userAssignedIdentities: { '${uami.id}': {} }
}
properties: {
encryption: {
keySource: 'Microsoft.KeyVault'
keyVaultProperties: [
{
keyName: 'eh-cmk'
keyVaultUri: kv.properties.vaultUri
identity: { userAssignedIdentity: uami.id }
}
]
requireInfrastructureEncryption: true // optional: double
}
}
}Then know how the key behaves in production:
- Rotation - new key versions are picked up automatically
- Key revoked - nothing is purged, but the namespace is inoperable until access is restored
- Checked - every 5 minutes
- Up to 3 keys - for redundancy
Protect it, and alert on the CustomerManagedKeyUserLogs.
Scenario 7: A bank needs a secure, highly available event platform across two regions. What do you set up?
Premium, because the bank needs geo-replication and customer-managed keys, and both need Premium or Dedicated. Each region is zone-redundant, and geo-replication adds the second region with events, metadata and offsets.
Security comes from everything we covered earlier:
- Microsoft Entra ID only -
disableLocalAuth: true - Managed identities - Data Sender / Data Receiver, narrow scope
- Private endpoints - public network access disabled
- Customer-managed key - Key Vault with purge protection
In Bicep: geo data replication with a primary and a secondary location, and a maximum replication lag of 5 minutes.
resource ns 'Microsoft.EventHub/namespaces@2026-01-01' = {
name: 'bank-events'
location: 'eastus2'
sku: { name: 'Premium', tier: 'Premium', capacity: 2 }
properties: {
disableLocalAuth: true // Microsoft Entra ID only
publicNetworkAccess: 'Disabled' // private endpoints only
geoDataReplication: {
maxReplicationLagDurationInSeconds: 300
locations: [
{ locationName: 'eastus2', roleType: 'Primary' }
{ locationName: 'centralus', roleType: 'Secondary' }
]
}
// + the encryption block from question six
}
}Then choose the mode. Synchronous gives RPO 0, but adds latency and ties availability to region two. Asynchronous allows a maximum lag, and past it the primary throttles. If the bank can't lose a single event, choose synchronous.
Then the drill. Promotion is always customer initiated, so practice it, and run a smoke test after every promotion. Ours sends a probe to each partition and reads it back. We run python smoke_test.py.
import logging, os, threading
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="payments")
with producer:
pids = producer.get_partition_ids()
start = {p: producer.get_partition_properties(p)
["last_enqueued_sequence_number"] for p in pids}
for p in pids: # one probe event per partition
producer.send_event(EventData(f"probe-{p}"), partition_id=p)
seen = {}
consumer = EventHubConsumerClient.from_connection_string(
conn, consumer_group="ops", eventhub_name="payments")
threading.Timer(8, consumer.close).start()
with consumer: # read only what came after start
consumer.receive(lambda pc, e: seen.update({pc.partition_id:
e.body_as_str()}), starting_position=start)
results = {p: seen.get(p) == f"probe-{p}" for p in pids}
for p, ok in results.items():
print(f"partition {p}:", "ok" if ok else "MISSING")
print(f"{sum(results.values())} of {len(pids)} partitions ok")$ python smoke_test.py
partition 0: ok
partition 1: ok
partition 2: ok
partition 3: ok
4 of 4 partitions okEvery partition prints ok, and the result is 4 of 4 partitions ok. Consumers must ignore probes.
Three details bite people:
- Checkpoint store - must be available in region 2
- Consumers + private endpoint - in region 2's VNet
- Retries - must ride out the promotion, typically under 2 minutes
One namespace host name, two regions, no secrets, no public access, and a drill that proves it works.
Key points to remember
- Microsoft Entra ID first; SAS only where needed, then disable local auth
- Managed identities + Data Sender / Receiver roles at the narrowest scope
- Private endpoint + public access disabled; get DNS right
- Zones are built in; send without partition info for availability
- Geo-DR = metadata only; Geo-replication = events and offsets too
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. One step restarts the emulator, and the script says so in a comment.
Show the full script that writes the files and runs every step
# ========================================================================
# Event Hubs Interview Q&A #9 - Security, networking and disaster recovery
# 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_09_Code.py.
# Entities in Config.json: orders (4 partitions, groups fraud); payments (4 partitions, groups ops)
# 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
from azure.storage.blob import BlobServiceClient
svc = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
svc.create_container("dr-checkpoints")
""")
run('python setup.py')
# Step 1: The parts of a SAS connection string, and the fields of a SAS token
write('sas_token.py', 'import base64, hashlib, hmac, os, urllib.parse\n\nconn = os.environ["EVENTHUB_CONNECTION_STRING"]\nparts = dict(p.split("=", 1) for p in conn.strip(";").split(";"))\nfor name, value in parts.items():\n shown = "*****" if name == "SharedAccessKey" else value\n print(f"{name:24} {shown}")\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\ntoken = sas_token("sb://localhost/orders", "orders-send", "SEND_KEY",\n 1798761600) # expiry: 2027-01-01 UTC\nfor field in token.split(" ", 1)[1].split("&"):\n print("token", field)\n')
run('python sas_token.py')
# Step 2: Sending with no partition information vs pinned to a partition ID
write('spread.py', """\
import os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="payments",
retry_total=5, retry_backoff_factor=0.8) # ride out a zone blip
def counts(): # events per partition
return [producer.get_partition_properties(p)
["last_enqueued_sequence_number"] + 1
for p in producer.get_partition_ids()]
with producer:
for i in range(8): # no key, no partition id
producer.send_event(EventData(f"payment-{i}"))
print("no partition info:", counts())
for i in range(8, 16): # pinned to one partition
producer.send_event(EventData(f"payment-{i}"), partition_id="2")
print("partition_id 2: ", counts())
""")
run('python spread.py')
# Step 3: The fraud consumer reads six orders and checkpoints
write('send.py', """\
import 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:
batch = producer.create_batch(partition_id="0")
for order_id in range(first, first + count):
batch.add(EventData(f"order-{order_id}"))
producer.send_batch(batch)
print(f"sent order-{first} to order-{first + count - 1}")
""")
write('consume.py', """\
import logging, os, threading
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"], "dr-checkpoints")
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], consumer_group="fraud",
eventhub_name="orders", checkpoint_store=store)
got = []
def on_event(pc, event):
got.append((pc.partition_id, event.sequence_number,
event.body_as_str()))
pc.update_checkpoint(event)
threading.Timer(8, client.close).start()
with client:
client.receive(on_event, starting_position="-1") # if no checkpoint
for p, seq, body in sorted(got):
print(f"read partition {p} seq {seq} {body}")
print("events read:", len(got))
for c in store.list_checkpoints("localhost", "orders", "fraud"):
print("checkpoint partition", c["partition_id"] + ":",
"seq", c["sequence_number"])
""")
run('python send.py 1001 6')
run('python consume.py')
# Step 4: Restart = same event hubs, no events (like a Geo-DR secondary): the old checkpoint reads nothing; reset it
# >>> now RESTART the emulator (the video restarts it here: all events are gone, checkpoints in Azurite stay) <<<
write('reset.py', """\
import os
from azure.storage.blob import ContainerClient
container = ContainerClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "dr-checkpoints")
prefix = "localhost/orders/fraud/checkpoint/"
for blob in container.list_blobs(name_starts_with=prefix):
container.delete_blob(blob.name)
print("deleted", blob.name)
""")
run('python send.py 2001 3')
run('python consume.py')
run('python reset.py')
run('python consume.py')
# Step 5: Smoke test after a promotion: a probe event through every partition
write('smoke_test.py', """\
import logging, os, threading
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="payments")
with producer:
pids = producer.get_partition_ids()
start = {p: producer.get_partition_properties(p)
["last_enqueued_sequence_number"] for p in pids}
for p in pids: # one probe event per partition
producer.send_event(EventData(f"probe-{p}"), partition_id=p)
seen = {}
consumer = EventHubConsumerClient.from_connection_string(
conn, consumer_group="ops", eventhub_name="payments")
threading.Timer(8, consumer.close).start()
with consumer: # read only what came after start
consumer.receive(lambda pc, e: seen.update({pc.partition_id:
e.body_as_str()}), starting_position=start)
results = {p: seen.get(p) == f"probe-{p}" for p in pids}
for p, ok in results.items():
print(f"partition {p}:", "ok" if ok else "MISSING")
print(f"{sum(results.values())} of {len(pids)} partitions ok")
""")
run('python smoke_test.py')
Part 10: Real-World Manager Scenarios
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.