In this post we are going to cover producers and sending events to 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 payments service must publish every transaction without losing any during an outage, and you build the producer.
A producer is any app that sends events to an event hub. Sending looks simple, one call, but most production problems start here: low throughput, events on the wrong partition, duplicates after a retry, payloads that are too big, and events lost when the service is down. Interviewers ask about producers because a good answer shows you know what happens between the send call and the acknowledgement.
Part 3 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: A producer sends one event per call and throughput is low. How does batching help?
- Question 2: Consumers need the event type without parsing the body. How do event properties work?
- Question 3: Someone asks whether to send to a partition key, a partition ID or neither. What is the difference?
- Question 4: A send fails during a network blip. How do retries work, and can they create duplicates?
- Question 5: A payload is too big to send. What are the size limits, and what do you do instead?
- Question 6: A teammate wants a fire-and-forget producer that batches for them. How does the buffered producer work?
- Scenario 7: A payments service must publish every transaction without losing any during an outage. How do you build the producer?
- The full script
The short answer
Send in batches: create_batch tracks the size limit and a whole batch is one request to one partition. Put routing data in event properties, use a partition key only when order matters, expect retries to create duplicates (Python has no idempotent producer), keep big payloads in blob storage, and never lose an event with an outbox and idempotent consumers.
Let's set up the project
We run Microsoft's official Event Hubs emulator locally, with Azurite for blob storage, and small Python scripts that use the azure-eventhub client, so every output is real. Features the emulator doesn't support are shown as code. The emulator reads its entities from Config.json: six event hubs, clicks, orders, deliveries, payments, invoices and transactions, each with 4 partitions. counts.py is a small helper that counts the events in each partition from the partition properties. Each step writes its own scripts.
{
"UserConfig": {
"NamespaceConfig": [
{
"Type": "EventHub",
"Name": "emulatorNs1",
"Entities": [
{
"Name": "clicks",
"PartitionCount": 4,
"ConsumerGroups": []
},
{
"Name": "orders",
"PartitionCount": 4,
"ConsumerGroups": []
},
{
"Name": "deliveries",
"PartitionCount": 4,
"ConsumerGroups": []
},
{
"Name": "payments",
"PartitionCount": 4,
"ConsumerGroups": []
},
{
"Name": "invoices",
"PartitionCount": 4,
"ConsumerGroups": []
},
{
"Name": "transactions",
"PartitionCount": 4,
"ConsumerGroups": []
}
]
}
],
"LoggingConfig": {
"Type": "File"
}
}
}def per_partition(producer):
"""Events in each partition, from the partition properties."""
counts = []
for pid in producer.get_partition_ids():
props = producer.get_partition_properties(pid)
counts.append(props["last_enqueued_sequence_number"] + 1)
return countsQuestion 1: A producer sends one event per call and throughput is low. How does batching help?
Batching packs many events into one send call. Every send waits for the service to acknowledge it, so one event per call means one round trip per event.
Let's measure it. send_one.py sends a thousand click events one by one. We run python send_one.py.
import json, os
from azure.eventhub import EventHubProducerClient, EventData
from counts import per_partition
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks")
calls = 0
with producer:
before = per_partition(producer)
for i in range(1000):
click = {"click_id": i, "page": f"/product/{i % 30}"}
producer.send_event(EventData(json.dumps(click))) # waits
calls += 1
after = per_partition(producer)
print("events: 1000, send calls:", calls)
print("events per partition:",
sorted(a - b for a, b in zip(after, before)))$ python send_one.py
events: 1000, send calls: 1000
events per partition: [250, 250, 250, 250]That's send calls: 1000, spread round robin over four partitions: [250, 250, 250, 250].
Now the same thousand clicks in a batch. send_batched.py calls create_batch, adds events until add raises a ValueError, then sends it and starts a new one. We run python send_batched.py.
import json, os
from azure.eventhub import EventHubProducerClient, EventData
from counts import per_partition
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks")
calls = 0
with producer:
before = per_partition(producer)
batch = producer.create_batch()
for i in range(1000):
click = {"click_id": i, "page": f"/product/{i % 30}"}
try:
batch.add(EventData(json.dumps(click)))
except ValueError: # full: send it, start anew
producer.send_batch(batch); calls += 1
batch = producer.create_batch()
batch.add(EventData(json.dumps(click)))
producer.send_batch(batch); calls += 1
after = per_partition(producer)
print("events: 1000, send calls:", calls, "| batch:",
batch.size_in_bytes, "of", batch.max_size_in_bytes, "bytes")
print("events per partition:",
sorted(a - b for a, b in zip(after, before)))$ python send_batched.py
events: 1000, send calls: 1 | batch: 49550 of 1048576 bytes
events per partition: [0, 0, 0, 1000]One send call: send calls: 1 | batch: 49550 of 1048576 bytes. The batch is under 50 KB of its 1 MB maximum.
Now look at the partition counts: [0, 0, 0, 1000]. A whole batch lands on one partition. And in Azure Monitor, a batch of 100 events counts as 1 incoming request and 100 incoming messages.
On Standard and Premium, the 1 MB limit covers the whole publish, batch included, not each event. So let create_batch track the size for you.
Question 2: Consumers need the event type without parsing the body. How do event properties work?
Every event has a body and a set of application properties: small key-value pairs that travel next to the body. That's where routing data belongs, like the event type and the schema version. There are also standard fields, such as content_type and message_id.
Let's try it. send_props.py gives each order event an event_type and a schema_version as properties, plus a content type and a message ID. Then read_props.py reads only the properties and routes each event. We run python send_props.py and python read_props.py.
import json, os
from azure.eventhub import EventHubProducerClient, EventData
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
events = [("OrderPlaced", 1001), ("PaymentFailed", 1002),
("OrderPlaced", 1003), ("OrderShipped", 1001)]
with producer:
batch = producer.create_batch(partition_id="0")
for event_type, order_id in events:
event = EventData(json.dumps({"order_id": order_id}))
event.properties = {"event_type": event_type,
"schema_version": 2}
event.content_type = "application/json"
event.message_id = f"{event_type}-{order_id}"
batch.add(event)
producer.send_batch(batch)
print("sent", len(events), "events with properties")import logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
handlers = {"OrderPlaced": "billing", "OrderShipped": "email"}
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="orders")
seen = []
def on_event(partition_context, event):
props = event.properties # keys come back as bytes
event_type = props[b"event_type"].decode()
target = handlers.get(event_type, "skipped") # body not parsed
seen.append((event.sequence_number, event_type, target,
props[b"schema_version"], event.content_type))
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, partition_id="0",
starting_position="-1")
for seq, event_type, target, version, ctype in sorted(seen):
print(seq, event_type, "->", target, "| v", version, ctype)$ python send_props.py
sent 4 events with properties
$ python read_props.py
0 OrderPlaced -> billing | v 2 application/json
1 PaymentFailed -> skipped | v 2 application/json
2 OrderPlaced -> billing | v 2 application/json
3 OrderShipped -> email | v 2 application/jsonOrderPlaced -> billing, OrderShipped -> email, and PaymentFailed -> skipped, because it has no handler here. The body is never parsed. Every event also shows v 2 application/json.
One detail with this Python client: property keys, and string values, come back as bytes. So we look up b"event_type" and decode it. Numbers keep their type: the schema version is still 2.
Keep the property contract small and agreed across teams. Properties are part of the event, so they count toward the size limit.
Question 3: Someone asks whether to send to a partition key, a partition ID or neither. What is the difference?
A partition key is hashed to pick a partition, so events with the same key stay together, in order. A partition ID pins the event to a partition we choose. With neither, events go round robin.
Let's send all three kinds to the deliveries hub. route.py sends four status events for order-1001 with a key, four for order-2002 pinned to partition 2, and eight clicks with neither. We run python route.py and python read_routes.py.
import os
from azure.eventhub import EventHubProducerClient, EventData
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="deliveries")
with producer:
for status in ["created", "packed", "shipped", "delivered"]:
producer.send_event(EventData(f"key order-1001 {status}"),
partition_key="order-1001") # hashed
for status in ["created", "packed", "shipped", "delivered"]:
producer.send_event(EventData(f"id order-2002 {status}"),
partition_id="2") # pinned
for i in range(8):
producer.send_event(EventData(f"none click-{i}")) # rotates
print("sent 4 with a key, 4 with a partition id, 8 with neither")import collections, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="deliveries")
got = collections.defaultdict(list)
def on_event(partition_context, event):
how, text = event.body_as_str().split(" ", 1)
got[how].append((partition_context.partition_id,
event.sequence_number, text))
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
for how in ["key", "id"]:
rows = sorted(got[how])
used = len({p for p, _, _ in rows})
print(f"partition_{how}: {len(rows)} events on {used} partition,",
"order:", [t.split()[1] for _, _, t in rows])
spread = collections.Counter(p for p, _, _ in got["none"])
print("neither: 8 events, per partition:", sorted(spread.values()))$ python route.py
sent 4 with a key, 4 with a partition id, 8 with neither
$ python read_routes.py
partition_key: 4 events on 1 partition, order: ['created', 'packed', 'shipped', 'delivered']
partition_id: 4 events on 1 partition, order: ['created', 'packed', 'shipped', 'delivered']
neither: 8 events, per partition: [2, 2, 2, 2]The keyed events: 4 events on 1 partition, in order, ['created', 'packed', 'shipped', 'delivered']. The pinned events are on one partition too, in the same order. And the eight clicks spread [2, 2, 2, 2].
| How you send | Where it goes | Order | Availability |
|---|---|---|---|
| partition_key="order-1001" | hashed: same key, same partition | kept per key | one partition |
| partition_id="2" | the partition you choose | kept in that partition | one partition |
| neither | round robin | none across events | skips an unavailable partition |
So which one? Neither, unless we need ordering. Microsoft's guidance is that a key or an ID downgrades availability to one partition: if it's unavailable, the send fails. Without them, events go to the remaining partitions.
Same key, same partition: that is the promise. Which partition is not: the emulator, the Python buffered producer and Kafka clients can map one key to different partitions.
Question 4: A send fails during a network blip. How do retries work, and can they create duplicates?
The client retries for us. Timeouts, throttling and a busy server are retried with exponential backoff. The Python defaults are three retries and a backoff factor of 0.8, so it waits 0.8, 1.6, then 3.2 seconds. An oversized event can never succeed, so it fails at once, with no retry.
Can retries create duplicates? Yes. Here is how:
- Send reaches the event hub - the event is stored
- The ack is lost - a network blip
- The client sends again - a second copy
Delivery is at least once. And the Python client has no idempotent producer, so the fix is on the consumer.
Let's simulate a resend. pay.py writes the retry defaults out, and sends pay-2002 twice with the same message ID, like an app that resent it after a timeout. Then dedupe.py counts message IDs. We run python pay.py and python dedupe.py.
import json, os
from azure.eventhub import EventHubProducerClient, EventData
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="payments",
retry_total=3, # the defaults, written out:
retry_backoff_factor=0.8, # waits 0.8 s, 1.6 s, 3.2 s
retry_backoff_max=120,
retry_mode="exponential")
# pay-2002 twice: the app resent it after a timeout with no ack
payments = ["pay-2001", "pay-2002", "pay-2002", "pay-2003"]
with producer:
for payment_id in payments:
event = EventData(json.dumps({"payment_id": payment_id}))
event.message_id = payment_id # same ID on every resend
producer.send_event(event)
print("sent", len(payments), "events")import collections, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="payments")
ids = collections.Counter()
def on_event(partition_context, event):
ids[event.message_id] += 1
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
print("events received:", sum(ids.values()))
print("unique payments:", len(ids))
print("duplicates:", sorted(k for k, n in ids.items() if n > 1))$ python pay.py
sent 4 events
$ python dedupe.py
events received: 4
unique payments: 3
duplicates: ['pay-2002']events received: 4, unique payments: 3, and duplicates: ['pay-2002']. The consumer found it.
The producer sets a business ID once and reuses it on every resend, and every consumer is idempotent on that ID. Kafka producers on Event Hubs do support idempotence.
Question 5: A payload is too big to send. What are the size limits, and what do you do instead?
The limit is per publication, a single event or a whole batch. Anything larger is rejected.
| Tier | Max publication (one event or a batch) |
|---|---|
| Basic | 256 KB |
| Standard | 1 MB |
| Premium | 1 MB |
| Dedicated | 20 MB (eligible clusters) |
The emulator has the 1 MB limit. Let's test it with python limits.py. It fills a batch with 1,000-byte events, asks for a 2 MB batch, then sends a 2 MB invoice PDF as one event.
import os, re
from azure.eventhub import EventHubProducerClient, EventData
from azure.eventhub.exceptions import EventHubError
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="invoices")
with producer:
batch = producer.create_batch()
print("batch max size:", batch.max_size_in_bytes, "bytes")
try:
while True:
batch.add(EventData("x" * 1000))
except ValueError as e:
print("full at", len(batch), "events of 1000 bytes:", e)
try:
producer.create_batch(max_size_in_bytes=2 * 1024 * 1024)
except ValueError as e:
print("ValueError:", e)
try:
producer.send_event(EventData(b"%PDF" + b"0" * 2_000_000))
except EventHubError as e:
first = str(e).split(" TrackingId")[0] # drop the run IDs
first = re.sub(r"delivery-id:\d+, ", "", first)
print(type(e).__name__ + ":", first)$ python limits.py
batch max size: 1048576 bytes
full at 1032 events of 1000 bytes: EventDataBatch has reached its size limit: 1048576
ValueError: Max message size: 2097152 is too large, acceptable max batch size is: 1048576 bytes.
EventDataSendError: The received message (size:1113534 bytes) exceeds the limit (1048576 bytes)
currently allowed on the link.batch max size: 1048576 bytes, and the batch is full at 1032 events of 1000 bytes. That's a client-side error, and so is asking for a 2 MB batch: ValueError: Max message size: 2097152 is too large. The PDF goes to the service, and the service rejects it: EventDataSendError, the message exceeds the limit (1048576 bytes) currently allowed on the link.
The fix is the claim check pattern. Upload the payload to blob storage, here Azurite, and send a small event with the blob path. We run python claim_check.py.
import json, os
from azure.eventhub import EventHubProducerClient, EventData
from azure.storage.blob import ContainerClient
pdf = b"%PDF" + b"0" * 2_000_000 # a 2 MB invoice
blobs = ContainerClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "invoices")
blobs.create_container()
blobs.upload_blob("2026/10/inv-5001.pdf", pdf) # 1. payload to blob
event = EventData(json.dumps({"invoice_id": "inv-5001",
"blob": "invoices/2026/10/inv-5001.pdf",
"bytes": len(pdf)}))
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="invoices")
with producer:
producer.send_event(event) # 2. small reference
print("blob uploaded:", len(pdf), "bytes")
print("event sent:", len(event.body_as_str()), "bytes")
print(" ", event.body_as_str())$ python claim_check.py
blob uploaded: 2000004 bytes
event sent: 85 bytes
{"invoice_id": "inv-5001", "blob": "invoices/2026/10/inv-5001.pdf", "bytes": 2000004}blob uploaded: 2000004 bytes, and event sent: 85 bytes, with the invoice ID and the blob path. The consumer downloads the PDF when it needs it.
For Kafka producers, Microsoft recommends max.request.size 1000000. 20 MB events need an eligible Dedicated cluster, and I still prefer the claim check.
Question 6: A teammate wants a fire-and-forget producer that batches for them. How does the buffered producer work?
Set buffered_mode=True. Then send_event returns at once: the client buffers per partition and publishes batches in the background. We must pass two callbacks, on_success and on_error, because that's where we learn the outcome.
Let's try it. buffered.py sends twelve clicks, tries one oversized event, then flushes. We run python buffered.py.
import os
from azure.eventhub import EventHubProducerClient, EventData
sent, failed = [], []
def on_success(events, partition_id):
sent.append((partition_id, len(events)))
def on_error(events, partition_id, error):
failed.append((partition_id, len(events), type(error).__name__))
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks",
buffered_mode=True, on_success=on_success, on_error=on_error,
max_wait_time=5, max_buffer_length=1500)
with producer:
for i in range(12):
producer.send_event(EventData(f"click-{i}")) # returns now
print("in the buffer:", producer.total_buffered_event_count)
try:
producer.send_event(EventData("x" * 1_100_000))
except ValueError as e:
print("raised at enqueue: ValueError:", e)
producer.flush()
print("after flush:", producer.total_buffered_event_count)
print("on_success (partition, events):", sorted(sent))
print("on_error:", failed)$ python buffered.py
in the buffer: 12
raised at enqueue: ValueError: EventDataBatch has reached its size limit: 1048576
after flush: 0
on_success (partition, events): [('0', 3), ('1', 3), ('2', 3), ('3', 3)]
on_error: []First, in the buffer: 12. The oversized event fails right away, when we enqueue it: raised at enqueue: ValueError. A failure after an event is in the buffer goes to on_error instead.
Then the flush sends the clicks, and we get after flush: 0. on_success is called once per partition, with three events each: [('0', 3), ('1', 3), ('2', 3), ('3', 3)]. And on_error: [].
When is a partition's buffer sent?
| Setting | Default |
|---|---|
| max_buffer_length | 1500 events per partition |
| max_wait_time | 1 second |
| close() | flushes the buffer |
A crash loses what's still in the buffer.
In buffered mode, the Python client picks the partition for a key itself, so don't mix it with plain producers when you rely on key order. And in on_error, log and resend; never drop events.
Scenario 7: A payments service must publish every transaction without losing any during an outage. How do you build the producer?
Four parts: a transactional outbox, client retries, a relay that marks events sent only after the acknowledgement, and an idempotent consumer.
Here is the flow:
- Payments DB - the payment and its outbox row, in one commit
- Relay - retries, then marks the rows sent
- transactions - the event hub, no key
- Consumer - dedupes on
message_id
So nothing is lost while Event Hubs is unreachable. The service sends the unsent outbox rows as one batch, with no partition key, so any healthy partition can take it.
Let's simulate an outage. With OUTAGE=1, pay_service.py points the producer at a port where nothing listens. We run it for txn-3001 and txn-3002, then run python pay_service.py txn-3003 with Event Hubs back, and python read_txn.py.
import json, os, sqlite3, sys
from azure.eventhub import EventHubProducerClient, EventData
from azure.eventhub.exceptions import EventHubError
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
if os.environ.get("OUTAGE"): # simulated: nothing listens here
conn = conn.replace("sb://localhost", "sb://localhost:5999")
db = sqlite3.connect("payments.db")
db.executescript("CREATE TABLE IF NOT EXISTS payments (txn_id TEXT"
" PRIMARY KEY, amount REAL); CREATE TABLE IF NOT EXISTS outbox"
" (txn_id TEXT PRIMARY KEY, body TEXT, sent INT DEFAULT 0);")
with db: # the payment and its event: one commit
for txn in sys.argv[1:]:
db.execute("INSERT INTO payments VALUES (?, 25.0)", (txn,))
db.execute("INSERT INTO outbox (txn_id, body) VALUES (?, ?)",
(txn, json.dumps({"txn_id": txn, "amount": 25.0})))
todo = db.execute("SELECT txn_id, body FROM outbox"
" WHERE sent = 0 ORDER BY txn_id").fetchall()
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name="transactions", retry_total=3)
try:
with producer:
batch = producer.create_batch() # no key: any partition
for txn_id, body in todo:
event = EventData(body); event.message_id = txn_id
batch.add(event)
producer.send_batch(batch) # returns after the ack
with db:
db.executemany("UPDATE outbox SET sent = 1 WHERE txn_id = ?",
[(t,) for t, _ in todo])
print("published from the outbox:", [t for t, _ in todo])
except EventHubError as e:
print("send failed after retries:", type(e).__name__)
print("kept in the outbox:", [t for t, _ in todo])import collections, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="transactions")
ids = collections.Counter()
def on_event(partition_context, event):
ids[event.message_id] += 1
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
print("events received:", sum(ids.values()))
print("unique transactions:", len(ids))
print("duplicates:", sorted(k for k, n in ids.items() if n > 1))$ OUTAGE=1 python pay_service.py txn-3001 txn-3002
send failed after retries: ConnectError
kept in the outbox: ['txn-3001', 'txn-3002']
$ python pay_service.py txn-3003
published from the outbox: ['txn-3001', 'txn-3002', 'txn-3003']
$ python read_txn.py
events received: 3
unique transactions: 3
duplicates: []During the outage: send failed after retries: ConnectError, and kept in the outbox: ['txn-3001', 'txn-3002']. When Event Hubs is back, the next run prints published from the outbox: ['txn-3001', 'txn-3002', 'txn-3003'], and only then marks them sent. The consumer sees events received: 3, unique transactions: 3, and duplicates: [].
If the service crashes after the ack but before marking the rows, they go out again. So the transaction ID is the message ID, and the consumer dedupes.
Availability zones are automatic where the region has them, and synchronous geo-replication on Premium and Dedicated covers a region outage.
Key points to remember
- Batch with create_batch: one request, one partition, 1 MB limit on Standard
- Routing data in event properties; Python returns keys as bytes
- No key unless order matters: a key or ID downgrades availability to one partition
- Retries can duplicate: no idempotent Python producer, so dedupe on a business ID
- Big payloads: claim check; never lose events: outbox + idempotent consumers
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 the Config.json above, run pip install azure-eventhub azure-eventhub-checkpointstoreblob azure-storage-blob confluent-kafka fastavro first, and set EVENTHUB_CONNECTION_STRING and STORAGE_CONNECTION_STRING.
Show the full script that writes the files and runs every step
# ========================================================================
# Event Hubs Interview Q&A #3 - Producers and sending events
# 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_03_Code.py.
# Entities in Config.json: clicks (4 partitions); orders (4 partitions); deliveries (4 partitions); payments (4 partitions); invoices (4 partitions); transactions (4 partitions)
# 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('counts.py', 'def per_partition(producer):\n """Events in each partition, from the partition properties."""\n counts = []\n for pid in producer.get_partition_ids():\n props = producer.get_partition_properties(pid)\n counts.append(props["last_enqueued_sequence_number"] + 1)\n return counts\n')
# Step 1: 1,000 click events sent one per call
write('send_one.py', """\
import json, os
from azure.eventhub import EventHubProducerClient, EventData
from counts import per_partition
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks")
calls = 0
with producer:
before = per_partition(producer)
for i in range(1000):
click = {"click_id": i, "page": f"/product/{i % 30}"}
producer.send_event(EventData(json.dumps(click))) # waits
calls += 1
after = per_partition(producer)
print("events: 1000, send calls:", calls)
print("events per partition:",
sorted(a - b for a, b in zip(after, before)))
""")
run('python send_one.py')
# Step 2: The same 1,000 click events in a batch
write('send_batched.py', """\
import json, os
from azure.eventhub import EventHubProducerClient, EventData
from counts import per_partition
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks")
calls = 0
with producer:
before = per_partition(producer)
batch = producer.create_batch()
for i in range(1000):
click = {"click_id": i, "page": f"/product/{i % 30}"}
try:
batch.add(EventData(json.dumps(click)))
except ValueError: # full: send it, start anew
producer.send_batch(batch); calls += 1
batch = producer.create_batch()
batch.add(EventData(json.dumps(click)))
producer.send_batch(batch); calls += 1
after = per_partition(producer)
print("events: 1000, send calls:", calls, "| batch:",
batch.size_in_bytes, "of", batch.max_size_in_bytes, "bytes")
print("events per partition:",
sorted(a - b for a, b in zip(after, before)))
""")
run('python send_batched.py')
# Step 3: Event properties set by the producer, routed by the consumer without parsing the body
write('send_props.py', """\
import json, os
from azure.eventhub import EventHubProducerClient, EventData
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
events = [("OrderPlaced", 1001), ("PaymentFailed", 1002),
("OrderPlaced", 1003), ("OrderShipped", 1001)]
with producer:
batch = producer.create_batch(partition_id="0")
for event_type, order_id in events:
event = EventData(json.dumps({"order_id": order_id}))
event.properties = {"event_type": event_type,
"schema_version": 2}
event.content_type = "application/json"
event.message_id = f"{event_type}-{order_id}"
batch.add(event)
producer.send_batch(batch)
print("sent", len(events), "events with properties")
""")
write('read_props.py', """\
import logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
handlers = {"OrderPlaced": "billing", "OrderShipped": "email"}
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="orders")
seen = []
def on_event(partition_context, event):
props = event.properties # keys come back as bytes
event_type = props[b"event_type"].decode()
target = handlers.get(event_type, "skipped") # body not parsed
seen.append((event.sequence_number, event_type, target,
props[b"schema_version"], event.content_type))
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, partition_id="0",
starting_position="-1")
for seq, event_type, target, version, ctype in sorted(seen):
print(seq, event_type, "->", target, "| v", version, ctype)
""")
run('python send_props.py')
run('python read_props.py')
# Step 4: Partition key, partition ID and neither (round robin)
write('route.py', """\
import os
from azure.eventhub import EventHubProducerClient, EventData
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="deliveries")
with producer:
for status in ["created", "packed", "shipped", "delivered"]:
producer.send_event(EventData(f"key order-1001 {status}"),
partition_key="order-1001") # hashed
for status in ["created", "packed", "shipped", "delivered"]:
producer.send_event(EventData(f"id order-2002 {status}"),
partition_id="2") # pinned
for i in range(8):
producer.send_event(EventData(f"none click-{i}")) # rotates
print("sent 4 with a key, 4 with a partition id, 8 with neither")
""")
write('read_routes.py', """\
import collections, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="deliveries")
got = collections.defaultdict(list)
def on_event(partition_context, event):
how, text = event.body_as_str().split(" ", 1)
got[how].append((partition_context.partition_id,
event.sequence_number, text))
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
for how in ["key", "id"]:
rows = sorted(got[how])
used = len({p for p, _, _ in rows})
print(f"partition_{how}: {len(rows)} events on {used} partition,",
"order:", [t.split()[1] for _, _, t in rows])
spread = collections.Counter(p for p, _, _ in got["none"])
print("neither: 8 events, per partition:", sorted(spread.values()))
""")
run('python route.py')
run('python read_routes.py')
# Step 5: Retry settings in the producer; a resent payment and a consumer that finds the duplicate
write('pay.py', """\
import json, os
from azure.eventhub import EventHubProducerClient, EventData
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="payments",
retry_total=3, # the defaults, written out:
retry_backoff_factor=0.8, # waits 0.8 s, 1.6 s, 3.2 s
retry_backoff_max=120,
retry_mode="exponential")
# pay-2002 twice: the app resent it after a timeout with no ack
payments = ["pay-2001", "pay-2002", "pay-2002", "pay-2003"]
with producer:
for payment_id in payments:
event = EventData(json.dumps({"payment_id": payment_id}))
event.message_id = payment_id # same ID on every resend
producer.send_event(event)
print("sent", len(payments), "events")
""")
write('dedupe.py', """\
import collections, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="payments")
ids = collections.Counter()
def on_event(partition_context, event):
ids[event.message_id] += 1
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
print("events received:", sum(ids.values()))
print("unique payments:", len(ids))
print("duplicates:", sorted(k for k, n in ids.items() if n > 1))
""")
run('python pay.py')
run('python dedupe.py')
# Step 6: The batch size limit and an oversized event
write('limits.py', 'import os, re\nfrom azure.eventhub import EventHubProducerClient, EventData\nfrom azure.eventhub.exceptions import EventHubError\n\nproducer = EventHubProducerClient.from_connection_string(\n os.environ["EVENTHUB_CONNECTION_STRING"],\n eventhub_name="invoices")\nwith producer:\n batch = producer.create_batch()\n print("batch max size:", batch.max_size_in_bytes, "bytes")\n try:\n while True:\n batch.add(EventData("x" * 1000))\n except ValueError as e:\n print("full at", len(batch), "events of 1000 bytes:", e)\n try:\n producer.create_batch(max_size_in_bytes=2 * 1024 * 1024)\n except ValueError as e:\n print("ValueError:", e)\n try:\n producer.send_event(EventData(b"%PDF" + b"0" * 2_000_000))\n except EventHubError as e:\n first = str(e).split(" TrackingId")[0] # drop the run IDs\n first = re.sub(r"delivery-id:\\d+, ", "", first)\n print(type(e).__name__ + ":", first)\n')
run('python limits.py')
# Step 7: The claim check pattern: payload to blob storage, a small reference event
write('claim_check.py', """\
import json, os
from azure.eventhub import EventHubProducerClient, EventData
from azure.storage.blob import ContainerClient
pdf = b"%PDF" + b"0" * 2_000_000 # a 2 MB invoice
blobs = ContainerClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "invoices")
blobs.create_container()
blobs.upload_blob("2026/10/inv-5001.pdf", pdf) # 1. payload to blob
event = EventData(json.dumps({"invoice_id": "inv-5001",
"blob": "invoices/2026/10/inv-5001.pdf",
"bytes": len(pdf)}))
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="invoices")
with producer:
producer.send_event(event) # 2. small reference
print("blob uploaded:", len(pdf), "bytes")
print("event sent:", len(event.body_as_str()), "bytes")
print(" ", event.body_as_str())
""")
run('python claim_check.py')
# Step 8: The buffered producer with on_success and on_error
write('buffered.py', """\
import os
from azure.eventhub import EventHubProducerClient, EventData
sent, failed = [], []
def on_success(events, partition_id):
sent.append((partition_id, len(events)))
def on_error(events, partition_id, error):
failed.append((partition_id, len(events), type(error).__name__))
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks",
buffered_mode=True, on_success=on_success, on_error=on_error,
max_wait_time=5, max_buffer_length=1500)
with producer:
for i in range(12):
producer.send_event(EventData(f"click-{i}")) # returns now
print("in the buffer:", producer.total_buffered_event_count)
try:
producer.send_event(EventData("x" * 1_100_000))
except ValueError as e:
print("raised at enqueue: ValueError:", e)
producer.flush()
print("after flush:", producer.total_buffered_event_count)
print("on_success (partition, events):", sorted(sent))
print("on_error:", failed)
""")
run('python buffered.py')
# Step 9: A payments producer with an outbox: an outage, then the relay publishes everything
write('pay_service.py', """\
import json, os, sqlite3, sys
from azure.eventhub import EventHubProducerClient, EventData
from azure.eventhub.exceptions import EventHubError
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
if os.environ.get("OUTAGE"): # simulated: nothing listens here
conn = conn.replace("sb://localhost", "sb://localhost:5999")
db = sqlite3.connect("payments.db")
db.executescript("CREATE TABLE IF NOT EXISTS payments (txn_id TEXT"
" PRIMARY KEY, amount REAL); CREATE TABLE IF NOT EXISTS outbox"
" (txn_id TEXT PRIMARY KEY, body TEXT, sent INT DEFAULT 0);")
with db: # the payment and its event: one commit
for txn in sys.argv[1:]:
db.execute("INSERT INTO payments VALUES (?, 25.0)", (txn,))
db.execute("INSERT INTO outbox (txn_id, body) VALUES (?, ?)",
(txn, json.dumps({"txn_id": txn, "amount": 25.0})))
todo = db.execute("SELECT txn_id, body FROM outbox"
" WHERE sent = 0 ORDER BY txn_id").fetchall()
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name="transactions", retry_total=3)
try:
with producer:
batch = producer.create_batch() # no key: any partition
for txn_id, body in todo:
event = EventData(body); event.message_id = txn_id
batch.add(event)
producer.send_batch(batch) # returns after the ack
with db:
db.executemany("UPDATE outbox SET sent = 1 WHERE txn_id = ?",
[(t,) for t, _ in todo])
print("published from the outbox:", [t for t, _ in todo])
except EventHubError as e:
print("send failed after retries:", type(e).__name__)
print("kept in the outbox:", [t for t, _ in todo])
""")
write('read_txn.py', """\
import collections, logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="$Default", eventhub_name="transactions")
ids = collections.Counter()
def on_event(partition_context, event):
ids[event.message_id] += 1
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
print("events received:", sum(ids.values()))
print("unique transactions:", len(ids))
print("duplicates:", sorted(k for k, n in ids.items() if n > 1))
""")
run('OUTAGE=1 python pay_service.py txn-3001 txn-3002')
run('python pay_service.py txn-3003')
run('python read_txn.py')
Part 4: Consumer Groups and Checkpointing
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.