In this post we are going to cover sessions and message ordering in Azure Service Bus. These are 7 real Service Bus interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: each order goes through placed, paid, packed and shipped, and the steps must never be processed out of order.
A session is a group of messages that share one SessionId. Service Bus hands the whole group to one receiver at a time, so those messages are processed in order. Interviewers ask about this because "keep it in order" sounds simple, and most people assume a queue already does it. It doesn't, and the wrong answer shows up in production as an old address overwriting a new one.
Part 6 of the Real Time Azure Service Bus Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: Updates for one customer arrive out of order. How do sessions guarantee order?
- Question 2: A teammate asks whether a plain queue keeps messages in order. What do you tell them?
- Question 3: Several consumers read a session-enabled queue. How are sessions shared between them?
- Question 4: A workflow needs to remember state between messages of one session. How does session state work?
- Question 5: One session gets most of the traffic and its consumer falls behind. How do you handle a hot session?
- Question 6: Someone asks what happens when a session consumer crashes. How does the session lock work?
- Scenario 7: Each order goes through placed, paid, packed and shipped, and steps must never be processed out of order. How do you design it?
- The full script
The short answer
A session groups messages by SessionId, and one receiver at a time holds its lock, so each session is processed first in, first out. A plain queue keeps queuing order, not processing order. Keep workflow progress in session state, key sessions by the smallest unit that must stay in order, and renew the session lock for long work.
Let's set up the project
We run Microsoft's official Service Bus emulator locally with SQL Server, and small Python scripts that use the azure-servicebus client, so every output is real. Features the emulator doesn't support are shown as code. Config.json creates one plain queue, customer-updates, and five queues that require sessions: customer-sessions, invoice-imports, marketplace, shipments with a 5 second lock duration, and order-workflow. common.py creates the client from the SERVICEBUS_CONNECTION_STRING environment variable and builds nine address updates: three customers, three updates each, interleaved.
{
"UserConfig": {
"Namespaces": [
{
"Name": "sbemulatorns",
"Queues": [
{
"Name": "customer-updates",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "customer-sessions",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "invoice-imports",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "marketplace",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "shipments",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT5S",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "order-workflow",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
}
],
"Topics": []
}
],
"Logging": {
"Type": "File"
}
}
}import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage
def client():
return ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
def updates(session=True):
"""Three customers, three address updates each, interleaved."""
return [ServiceBusMessage(
f"{c}{n}", message_id=f"{c}{n}",
session_id=f"cust-{c}" if session else None)
for n in (1, 2, 3) for c in "ABC"]Question 1: Updates for one customer arrive out of order. How do sessions guarantee order?
A session groups messages by session ID, and a session receiver locks the whole session. One receiver holds it at a time and gets its messages in order. So with the customer as the session ID, each customer's updates are processed first in, first out.
Let's try it. The customer-sessions queue has RequiresSession set to true. Updates for customers A, B and C arrive interleaved, A1 B1 C1 A2 B2 C2 A3 B3 C3, each with the customer as the session ID. Then one session receiver per customer reads its own session. We run python sessions.py.
from common import client, updates
with client() as sb:
with sb.get_queue_sender("customer-sessions") as sender:
sender.send_messages(updates()) # A1 B1 C1 A2 B2 C2 ...
for customer in ["cust-A", "cust-B", "cust-C"]:
with sb.get_queue_receiver("customer-sessions",
session_id=customer) as r:
got = r.receive_messages(max_message_count=10,
max_wait_time=3)
for msg in got:
r.complete_message(msg)
print(customer, "->", [f"{m} (seq {m.sequence_number})"
for m in got])$ python sessions.py
cust-A -> ['A1 (seq 1)', 'A2 (seq 4)', 'A3 (seq 7)']
cust-B -> ['B1 (seq 2)', 'B2 (seq 5)', 'B3 (seq 8)']
cust-C -> ['C1 (seq 3)', 'C2 (seq 6)', 'C3 (seq 9)']Customer A gets A1 (seq 1), A2 (seq 4) and A3 (seq 7): its own updates, in order, even though the queue holds all nine. Same for B, with sequence numbers 2, 5 and 8, and for C, with 3, 6 and 9.
Before you design with sessions, check two things:
| Sessions | |
|---|---|
| Basic tier | not supported |
| Standard, Premium | supported |
| Turn on later? | no: set at creation |
Basic doesn't have sessions, and you can't turn them on or off after the queue or subscription is created. So this is a design-time decision.
Pick the session ID from what must stay in order, like a customer or an order. There's no order across sessions, and that is exactly what lets you scale.
Question 2: A teammate asks whether a plain queue keeps messages in order. What do you tell them?
It stores them in order, but it doesn't guarantee processing order. Sequence numbers follow the queuing order, and receivers usually get messages in that order. But Microsoft is clear: processing order requires sessions.
There are three ways order breaks on a plain queue:
- Competing consumers - no order across workers
- Abandon or lock expiry - the retry lands later
- Prefetch - can deliver out of order
Let's see the first two. compete.py sends the same nine updates to the plain customer-updates queue, and two workers take turns. Then retry.py sends A1, A2 and A3, and A1 fails once with a transient error and is abandoned. We run python compete.py and python retry.py.
from common import client, updates
with client() as sb:
with sb.get_queue_sender("customer-updates") as sender:
sender.send_messages(updates(session=False))
workers = {"worker-1": [], "worker-2": []}
receivers = {w: sb.get_queue_receiver("customer-updates")
for w in workers}
for _ in range(5): # the workers take turns
for w, r in receivers.items():
for msg in r.receive_messages(max_message_count=1,
max_wait_time=3):
workers[w].append(str(msg))
r.complete_message(msg)
for w, r in receivers.items():
r.close()
print(w, "got:", " ".join(workers[w]))from azure.servicebus import ServiceBusMessage
from common import client
applied = []
with client() as sb:
with sb.get_queue_sender("customer-updates") as sender:
sender.send_messages([ServiceBusMessage(f"A{n}")
for n in (1, 2, 3)])
with sb.get_queue_receiver("customer-updates") as r:
while batch := r.receive_messages(max_message_count=10,
max_wait_time=3):
for msg in batch:
if str(msg) == "A1" and msg.delivery_count == 0:
r.abandon_message(msg) # a transient error
continue
applied.append(str(msg))
r.complete_message(msg)
print("applied in this order:", " ".join(applied))
print("final address comes from:", applied[-1])$ python compete.py
worker-1 got: A1 C1 B2 A3 C3
worker-2 got: B1 A2 C2 B3
$ python retry.py
applied in this order: A2 A3 A1
final address comes from: A1Customer A's updates are split: worker-1 got: A1 C1 B2 A3 C3 and worker-2 got: B1 A2 C2 B3. If worker two is faster, A2 is applied before A1.
The retry run prints applied in this order: A2 A3 A1. A2 and A3 from the same batch were applied, then A1 came back and was applied last. So final address comes from: A1, the oldest update.
Prefetch adds to it. So with competing consumers, assume no order at all.
If order matters only sometimes, make the consumer tolerant: carry a version in each message and skip stale ones. When it always matters, use sessions, and never process past a message that failed.
Question 3: Several consumers read a session-enabled queue. How are sessions shared between them?
Each consumer asks for the next available session. Service Bus hands it a session that no one else holds and locks it for that consumer. So consumers share the queue session by session, never message by message within one session.
Let's run it. share.py sends the nine updates to customer-sessions, then opens two consumers at once, both with NEXT_AVAILABLE_SESSION. When they close, consumer one asks again, and then consumer two. We run python share.py.
from azure.servicebus import NEXT_AVAILABLE_SESSION
from azure.servicebus.exceptions import OperationTimeoutError
from common import client, updates
def consumer(sb): # locks the next free session
return sb.get_queue_receiver("customer-sessions", max_wait_time=3,
session_id=NEXT_AVAILABLE_SESSION)
def drain(name, r):
got = r.receive_messages(max_message_count=10, max_wait_time=3)
for msg in got:
r.complete_message(msg)
print(name, "locked", r.session.session_id, "->",
" ".join(str(m) for m in got))
with client() as sb:
with sb.get_queue_sender("customer-sessions") as sender:
sender.send_messages(updates())
with consumer(sb) as c1, consumer(sb) as c2: # both at once
drain("consumer-1", c1)
drain("consumer-2", c2)
with consumer(sb) as c1: # A and B released
drain("consumer-1", c1)
try:
with consumer(sb) as c2:
drain("consumer-2", c2)
except OperationTimeoutError as e:
print("consumer-2:", type(e).__name__, "(no free session)")$ python share.py
consumer-1 locked cust-A -> A1 A2 A3
consumer-2 locked cust-B -> B1 B2 B3
consumer-1 locked cust-C -> C1 C2 C3
consumer-2: OperationTimeoutError (no free session)Consumer one gets cust-A and reads A1 A2 A3. Consumer two gets cust-B, because A is already locked. Consumer one finishes A and closes, which releases the session, then asks again and gets cust-C. When nothing is left, consumer two's request times out with OperationTimeoutError. A real consumer simply tries again later.
In Azure Functions, the trigger does this for you. maxConcurrentSessions in host.json, 8 by default, sets how many sessions it handles at once:
{
"version": "2.0",
"extensions": {
"serviceBus": {
"maxConcurrentSessions": 8
}
}
}The emulator doesn't enforce the exclusive session lock, so a second receiver can take a session that's already locked. That's why our demos use one receiver per session. On Azure, only one receiver can hold a session's lock.
Question 4: A workflow needs to remember state between messages of one session. How does session state work?
Each session can keep state on the broker. It is an opaque binary value up to the size of one message: 256 KB on Standard, 100 MB on Premium. It stays until you clear it, even after all messages are consumed.
Let's build an invoice import. Invoice inv-9001 arrives in three parts, all in one session. The importer reads the state, adds each part to a running total, saves the state, and only then completes the message. The first run stops after two parts, like a crash or a deploy. Then a second run picks up where it left off. We run python send_parts.py, python import_parts.py 2 and python import_parts.py.
import json
from azure.servicebus import ServiceBusMessage
from common import client
parts = [{"part": 1, "amount": 100}, {"part": 2, "amount": 200},
{"part": 3, "amount": 150}]
with client() as sb, sb.get_queue_sender("invoice-imports") as s:
s.send_messages([ServiceBusMessage(json.dumps(p),
session_id="inv-9001")
for p in parts])
print("sent 3 parts of invoice inv-9001")import json, sys
from common import client
limit = int(sys.argv[1]) if len(sys.argv) > 1 else 10 # stop early
with client() as sb, sb.get_queue_receiver(
"invoice-imports", session_id="inv-9001") as r:
raw = r.session.get_state() # bytes, or None
print("state at start:", raw)
state = json.loads(raw) if raw else {"parts": 0, "total": 0,
"last_seq": 0}
for msg in r.receive_messages(max_message_count=limit,
max_wait_time=3):
if msg.sequence_number > state["last_seq"]: # not seen yet
state["parts"] += 1
state["total"] += json.loads(str(msg))["amount"]
state["last_seq"] = msg.sequence_number
r.session.set_state(json.dumps(state)) # save first
r.complete_message(msg) # then settle
if state["parts"] == 3:
print("invoice imported:", state)
r.session.set_state(None) # clean up
print("state at end:", r.session.get_state())$ python send_parts.py
sent 3 parts of invoice inv-9001
$ python import_parts.py 2
state at start: None
state at end: b'{"parts": 2, "total": 300, "last_seq": 2}'
$ python import_parts.py
state at start: b'{"parts": 2, "total": 300, "last_seq": 2}'
invoice imported: {'parts': 3, 'total': 450, 'last_seq': 3}
state at end: NoneThe first run prints state at start: None and ends with {"parts": 2, "total": 300, "last_seq": 2}. It comes back as bytes, b'...', so decode it before you use it.
The second run starts from that state, adds part three, and prints invoice imported: {'parts': 3, 'total': 450, 'last_seq': 3}. Then set_state(None) clears it, so state at end: None.
The last_seq field makes it idempotent. If the process dies after saving the state but before completing the message, the redelivered part has a sequence number we already counted, so it is skipped.
State stays until you clear it. Always clear it when the workflow is done.
Question 5: One session gets most of the traffic and its consumer falls behind. How do you handle a hot session?
A session is processed by one receiver at a time, so one session can only go as fast as one consumer. More consumers or a bigger tier won't parallelize it. The fix is the session key.
Let's measure it. hot.py sends eighty marketplace events, placed and paid. One big shop has thirty orders, and five small shops have two each. First we key the sessions by seller, then by order, and each time we drain the queue session by session and count. We run python hot.py seller and python hot.py order.
import collections, sys
from azure.servicebus import ServiceBusMessage, NEXT_AVAILABLE_SESSION
from azure.servicebus.exceptions import OperationTimeoutError
from common import client
key = sys.argv[1] # "seller" or "order"
orders = [("big-shop", n) for n in range(30)] + [
(f"shop-{s}", n) for s in "abcde" for n in (1, 2)]
counts = collections.Counter()
with client() as sb:
with sb.get_queue_sender("marketplace") as sender:
sender.send_messages([ServiceBusMessage(
event, session_id=seller if key == "seller"
else f"{seller}-order-{n}")
for seller, n in orders for event in ("placed", "paid")])
while True: # one session at a time
try:
with sb.get_queue_receiver(
"marketplace", max_wait_time=3,
session_id=NEXT_AVAILABLE_SESSION) as r:
for msg in r.receive_messages(max_message_count=100,
max_wait_time=3):
r.complete_message(msg)
counts[r.session.session_id] += 1
except OperationTimeoutError:
break
print(f"by {key}: {len(counts)} sessions, messages per session:",
sorted(counts.values(), reverse=True)[:8])$ python hot.py seller
by seller: 6 sessions, messages per session: [60, 4, 4, 4, 4, 4]
$ python hot.py order
by order: 40 sessions, messages per session: [2, 2, 2, 2, 2, 2, 2, 2]By seller we get 6 sessions, and one of them holds 60 of the eighty messages. The other five hold 4 each. While one consumer works through the big shop, the others finish their four messages and sit idle.
By order we get 40 sessions, two messages each. Placed still comes before paid for each order, and forty sessions spread across all consumers.
| Session key | Sessions | Largest session |
|---|---|---|
| seller | 6 | 60 of 80 |
| order | 40 | 2 |
So key by the smallest unit that must stay in order: the order, not the seller or the region.
One consumer always busy while the others sit idle is the classic sign of a hot session.
Question 6: Someone asks what happens when a session consumer crashes. How does the session lock work?
The session lock is an umbrella over the message locks. It times out unless you renew it. When the consumer closes the session, or the lock expires, the session is released for another consumer.
Let's crash one. The shipments queue has a 5 second lock duration. Consumer one receives label-printed and hangs for seven seconds, longer than the lock, then tries to complete. Consumer two then takes the session, registers it with an AutoLockRenewer, works for eight seconds and completes. We run python lock.py.
import time
from azure.servicebus import AutoLockRenewer, ServiceBusMessage
from azure.servicebus.exceptions import SessionLockLostError
from common import client
with client() as sb: # shipments: LockDuration 5 s
with sb.get_queue_sender("shipments") as s:
s.send_messages([ServiceBusMessage(e, session_id="ship-77")
for e in ("label-printed", "picked-up")])
with sb.get_queue_receiver("shipments", session_id="ship-77",
max_wait_time=3) as r: # consumer 1
msg = r.receive_messages(max_wait_time=3)[0]
print("consumer 1 got", msg, "delivery_count",
msg.delivery_count)
time.sleep(7) # hangs: longer than 5 s
try:
r.complete_message(msg)
except SessionLockLostError as e:
print("consumer 1:", type(e).__name__, "|",
str(e).split(". ")[0])
with sb.get_queue_receiver("shipments", session_id="ship-77",
max_wait_time=3) as r, AutoLockRenewer() as renewer:
renewer.register(r, r.session, max_lock_renewal_duration=60)
got = r.receive_messages(max_message_count=10,
max_wait_time=3)
print("consumer 2 got",
[(str(m), m.delivery_count) for m in got])
time.sleep(8) # slow work, renewed
for m in got:
r.complete_message(m)
print("consumer 2 completed", len(got), "after 8 s of work")$ python lock.py
consumer 1 got label-printed delivery_count 0
consumer 1: SessionLockLostError | The lock on the session has expired
consumer 2 got [('label-printed', 1), ('picked-up', 0)]
consumer 2 completed 2 after 8 s of workConsumer one got label-printed with delivery_count 0, zero on the first try. When it tries to complete, it gets SessionLockLostError: The lock on the session has expired.
Consumer two gets ('label-printed', 1) and ('picked-up', 0). The delivery count went up because the lock timed out. With the renewer, eight seconds of work is fine, and it prints consumer 2 completed 2 after 8 s of work.
Microsoft's rule on the delivery count:
| What happens | Delivery count |
|---|---|
| the session lock expires | goes up |
| closed without completing | unchanged |
And messages in a session don't need their own lock renewals: you renew the session.
Renew the lock for long work, close the session as soon as you're done, and stay idempotent: after a lock loss, the same message comes again.
Scenario 7: Each order goes through placed, paid, packed and shipped, and steps must never be processed out of order. How do you design it?
A session per order. The session ID is the order ID, so each order's events, placed, paid, packed and shipped, reach one consumer at a time, in order, while orders run in parallel.
First, create the queue with sessions enabled, because you can't turn them on later:
# Azure: sessions can't be turned on later
az servicebus queue create --resource-group rg-shop \
--namespace-name shop-sb --name order-workflow \
--enable-session trueThen the consumer keeps a small state machine in session state: the steps done, plus any step that came too early. It checks each event against the next expected step.
Let's run it. send_workflow.py sends ten events for three orders, with a publisher bug: shipped goes out before packed for order-503. Sessions deliver what was sent, in order, so the consumer defers shipped, keeps its sequence number in state, and after packed it receives the deferred message. We run python send_workflow.py and python workflow.py.
from azure.servicebus import ServiceBusMessage
from common import client
events = [("order-501", "placed"), ("order-502", "placed"),
("order-501", "paid"), ("order-503", "placed"),
("order-502", "paid"), ("order-501", "packed"),
("order-503", "paid"), ("order-503", "shipped"), # early!
("order-501", "shipped"), ("order-503", "packed")]
with client() as sb, sb.get_queue_sender("order-workflow") as s:
s.send_messages([ServiceBusMessage(f"{o} {step}", session_id=o,
subject=step)
for o, step in events])
print("sent", len(events), "events for 3 orders")import json
STEPS = ["placed", "paid", "packed", "shipped"]
EMPTY = '{"done": [], "deferred": {}}'
def save(r, state):
r.session.set_state(json.dumps(state))
def handle(r, msg, state, log):
step, done = msg.subject, state["done"]
if step in done: # redelivered
log.append(f"{step}: duplicate, skipped")
elif step != STEPS[len(done)]: # too early
state["deferred"][step] = msg.sequence_number
save(r, state) # remember it
r.defer_message(msg)
log.append(f"{step}: deferred")
return
else:
done.append(step) # do the work
state["deferred"].pop(step, None)
log.append(step)
save(r, state) # then settle
r.complete_message(msg)
nxt = STEPS[len(done)] if len(done) < len(STEPS) else None
if nxt in state["deferred"]: # its turn now
seq = state["deferred"][nxt]
handle(r, r.receive_deferred_messages([seq])[0], state, log)import json
from azure.servicebus import NEXT_AVAILABLE_SESSION
from azure.servicebus.exceptions import OperationTimeoutError
from common import client
from steps import EMPTY, handle
with client() as sb:
while True: # one order at a time
try:
with sb.get_queue_receiver(
"order-workflow", max_wait_time=3,
session_id=NEXT_AVAILABLE_SESSION) as r:
state = json.loads(r.session.get_state() or EMPTY)
log = []
for msg in r.receive_messages(max_message_count=20,
max_wait_time=3):
handle(r, msg, state, log)
if len(state["done"]) == 4: # finished: clean up
r.session.set_state(None)
print(r.session.session_id, "|", ", ".join(log))
print(" state:", r.session.get_state())
except OperationTimeoutError:
break$ python send_workflow.py
sent 10 events for 3 orders
$ python workflow.py
order-501 | placed, paid, packed, shipped
state: None
order-502 | placed, paid
state: b'{"done": ["placed", "paid"], "deferred": {}}'
order-503 | placed, paid, shipped: deferred, packed, shipped
state: Noneorder-501 runs placed, paid, packed, shipped, then clears its state: state: None. order-502 stops at placed, paid, with its state saved for the next event. order-503 shows placed, paid, shipped: deferred, packed, shipped: the early step waited for its turn. A redelivered step would be skipped, because it is already in done.
Three rules keep it safe:
- Save state, then settle
- One session per order - not per customer, so no session gets hot
- Alert on dead letters - orders that never complete
Each order is strictly in order, all orders run in parallel, and an early or repeated step can't break the workflow.
Key points to remember
- A session = messages with one SessionId; one receiver holds its lock, in order
- A plain queue keeps queuing order, not processing order
- Consumers take the next available session; Functions: maxConcurrentSessions 8
- Session state: up to one message in size, kept until you clear it
- Key by the smallest ordered unit; renew the session lock for long work
The full script
This is a Python script that writes every file and runs every command above in order against the Service Bus emulator: start the emulator with SQL Server, run pip install azure-servicebus first, set SERVICEBUS_CONNECTION_STRING, and run it in an empty folder.
Show the full script that writes the files and runs every step
# ========================================================================
# Service Bus Interview Q&A #6 - Sessions and message ordering
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Azure Service Bus emulator (Docker image, version 2.0.1) with
# SQL Server 2022 and azure-servicebus 7.15.0; 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-service-bus-emulator-installer),
# using the Config.json this script writes first (Step 0) as the emulator's config file; it holds every
# queue, topic, subscription, rule and setting (lock duration, max delivery count, duplicate detection...)
# the steps depend on.
# 2. pip install azure-servicebus
# 3. Set SERVICEBUS_CONNECTION_STRING to the emulator connection string (Endpoint=sb://localhost;
# SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;)
# then run python ServiceBus_Interview_QA_06_Code.py.
# Run it in an empty folder and on a freshly started emulator: it creates its files (and any SQLite
# databases) there, and a second run on old files or old messages gives different output.
# 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 emulator config (mount it as ServiceBus_Emulator/ConfigFiles/Config.json, then start the emulator)
write('Config.json', """\
{
"UserConfig": {
"Namespaces": [
{
"Name": "sbemulatorns",
"Queues": [
{
"Name": "customer-updates",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "customer-sessions",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "invoice-imports",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "marketplace",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "shipments",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT5S",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
},
{
"Name": "order-workflow",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": true
}
}
],
"Topics": []
}
],
"Logging": {
"Type": "File"
}
}
}
""")
# Step 0: the starting files
write('common.py', 'import os\nfrom azure.servicebus import ServiceBusClient, ServiceBusMessage\n\n\ndef client():\n return ServiceBusClient.from_connection_string(\n os.environ["SERVICEBUS_CONNECTION_STRING"])\n\n\ndef updates(session=True):\n """Three customers, three address updates each, interleaved."""\n return [ServiceBusMessage(\n f"{c}{n}", message_id=f"{c}{n}",\n session_id=f"cust-{c}" if session else None)\n for n in (1, 2, 3) for c in "ABC"]\n')
# Step 1: Interleaved updates for three customers: one session receiver per customer
write('sessions.py', """\
from common import client, updates
with client() as sb:
with sb.get_queue_sender("customer-sessions") as sender:
sender.send_messages(updates()) # A1 B1 C1 A2 B2 C2 ...
for customer in ["cust-A", "cust-B", "cust-C"]:
with sb.get_queue_receiver("customer-sessions",
session_id=customer) as r:
got = r.receive_messages(max_message_count=10,
max_wait_time=3)
for msg in got:
r.complete_message(msg)
print(customer, "->", [f"{m} (seq {m.sequence_number})"
for m in got])
""")
run('python sessions.py')
# Step 2: A plain queue: two competing workers, and an abandon that reorders
write('compete.py', """\
from common import client, updates
with client() as sb:
with sb.get_queue_sender("customer-updates") as sender:
sender.send_messages(updates(session=False))
workers = {"worker-1": [], "worker-2": []}
receivers = {w: sb.get_queue_receiver("customer-updates")
for w in workers}
for _ in range(5): # the workers take turns
for w, r in receivers.items():
for msg in r.receive_messages(max_message_count=1,
max_wait_time=3):
workers[w].append(str(msg))
r.complete_message(msg)
for w, r in receivers.items():
r.close()
print(w, "got:", " ".join(workers[w]))
""")
write('retry.py', """\
from azure.servicebus import ServiceBusMessage
from common import client
applied = []
with client() as sb:
with sb.get_queue_sender("customer-updates") as sender:
sender.send_messages([ServiceBusMessage(f"A{n}")
for n in (1, 2, 3)])
with sb.get_queue_receiver("customer-updates") as r:
while batch := r.receive_messages(max_message_count=10,
max_wait_time=3):
for msg in batch:
if str(msg) == "A1" and msg.delivery_count == 0:
r.abandon_message(msg) # a transient error
continue
applied.append(str(msg))
r.complete_message(msg)
print("applied in this order:", " ".join(applied))
print("final address comes from:", applied[-1])
""")
run('python compete.py')
run('python retry.py')
# Step 3: Two consumers share a session queue with NEXT_AVAILABLE_SESSION
write('share.py', """\
from azure.servicebus import NEXT_AVAILABLE_SESSION
from azure.servicebus.exceptions import OperationTimeoutError
from common import client, updates
def consumer(sb): # locks the next free session
return sb.get_queue_receiver("customer-sessions", max_wait_time=3,
session_id=NEXT_AVAILABLE_SESSION)
def drain(name, r):
got = r.receive_messages(max_message_count=10, max_wait_time=3)
for msg in got:
r.complete_message(msg)
print(name, "locked", r.session.session_id, "->",
" ".join(str(m) for m in got))
with client() as sb:
with sb.get_queue_sender("customer-sessions") as sender:
sender.send_messages(updates())
with consumer(sb) as c1, consumer(sb) as c2: # both at once
drain("consumer-1", c1)
drain("consumer-2", c2)
with consumer(sb) as c1: # A and B released
drain("consumer-1", c1)
try:
with consumer(sb) as c2:
drain("consumer-2", c2)
except OperationTimeoutError as e:
print("consumer-2:", type(e).__name__, "(no free session)")
""")
run('python share.py')
# Step 4: Session state: an invoice import that stops after two parts and resumes
write('send_parts.py', """\
import json
from azure.servicebus import ServiceBusMessage
from common import client
parts = [{"part": 1, "amount": 100}, {"part": 2, "amount": 200},
{"part": 3, "amount": 150}]
with client() as sb, sb.get_queue_sender("invoice-imports") as s:
s.send_messages([ServiceBusMessage(json.dumps(p),
session_id="inv-9001")
for p in parts])
print("sent 3 parts of invoice inv-9001")
""")
write('import_parts.py', """\
import json, sys
from common import client
limit = int(sys.argv[1]) if len(sys.argv) > 1 else 10 # stop early
with client() as sb, sb.get_queue_receiver(
"invoice-imports", session_id="inv-9001") as r:
raw = r.session.get_state() # bytes, or None
print("state at start:", raw)
state = json.loads(raw) if raw else {"parts": 0, "total": 0,
"last_seq": 0}
for msg in r.receive_messages(max_message_count=limit,
max_wait_time=3):
if msg.sequence_number > state["last_seq"]: # not seen yet
state["parts"] += 1
state["total"] += json.loads(str(msg))["amount"]
state["last_seq"] = msg.sequence_number
r.session.set_state(json.dumps(state)) # save first
r.complete_message(msg) # then settle
if state["parts"] == 3:
print("invoice imported:", state)
r.session.set_state(None) # clean up
print("state at end:", r.session.get_state())
""")
run('python send_parts.py')
run('python import_parts.py 2')
run('python import_parts.py')
# Step 5: A hot session: keyed by seller vs keyed by order
write('hot.py', """\
import collections, sys
from azure.servicebus import ServiceBusMessage, NEXT_AVAILABLE_SESSION
from azure.servicebus.exceptions import OperationTimeoutError
from common import client
key = sys.argv[1] # "seller" or "order"
orders = [("big-shop", n) for n in range(30)] + [
(f"shop-{s}", n) for s in "abcde" for n in (1, 2)]
counts = collections.Counter()
with client() as sb:
with sb.get_queue_sender("marketplace") as sender:
sender.send_messages([ServiceBusMessage(
event, session_id=seller if key == "seller"
else f"{seller}-order-{n}")
for seller, n in orders for event in ("placed", "paid")])
while True: # one session at a time
try:
with sb.get_queue_receiver(
"marketplace", max_wait_time=3,
session_id=NEXT_AVAILABLE_SESSION) as r:
for msg in r.receive_messages(max_message_count=100,
max_wait_time=3):
r.complete_message(msg)
counts[r.session.session_id] += 1
except OperationTimeoutError:
break
print(f"by {key}: {len(counts)} sessions, messages per session:",
sorted(counts.values(), reverse=True)[:8])
""")
run('python hot.py seller')
run('python hot.py order')
# Step 6: Session lock: a hung consumer loses the session; renewal keeps it
write('lock.py', """\
import time
from azure.servicebus import AutoLockRenewer, ServiceBusMessage
from azure.servicebus.exceptions import SessionLockLostError
from common import client
with client() as sb: # shipments: LockDuration 5 s
with sb.get_queue_sender("shipments") as s:
s.send_messages([ServiceBusMessage(e, session_id="ship-77")
for e in ("label-printed", "picked-up")])
with sb.get_queue_receiver("shipments", session_id="ship-77",
max_wait_time=3) as r: # consumer 1
msg = r.receive_messages(max_wait_time=3)[0]
print("consumer 1 got", msg, "delivery_count",
msg.delivery_count)
time.sleep(7) # hangs: longer than 5 s
try:
r.complete_message(msg)
except SessionLockLostError as e:
print("consumer 1:", type(e).__name__, "|",
str(e).split(". ")[0])
with sb.get_queue_receiver("shipments", session_id="ship-77",
max_wait_time=3) as r, AutoLockRenewer() as renewer:
renewer.register(r, r.session, max_lock_renewal_duration=60)
got = r.receive_messages(max_message_count=10,
max_wait_time=3)
print("consumer 2 got",
[(str(m), m.delivery_count) for m in got])
time.sleep(8) # slow work, renewed
for m in got:
r.complete_message(m)
print("consumer 2 completed", len(got), "after 8 s of work")
""")
run('python lock.py')
# Step 7: placed, paid, packed, shipped: session per order, state, and deferral of an early step
write('send_workflow.py', """\
from azure.servicebus import ServiceBusMessage
from common import client
events = [("order-501", "placed"), ("order-502", "placed"),
("order-501", "paid"), ("order-503", "placed"),
("order-502", "paid"), ("order-501", "packed"),
("order-503", "paid"), ("order-503", "shipped"), # early!
("order-501", "shipped"), ("order-503", "packed")]
with client() as sb, sb.get_queue_sender("order-workflow") as s:
s.send_messages([ServiceBusMessage(f"{o} {step}", session_id=o,
subject=step)
for o, step in events])
print("sent", len(events), "events for 3 orders")
""")
write('steps.py', """\
import json
STEPS = ["placed", "paid", "packed", "shipped"]
EMPTY = '{"done": [], "deferred": {}}'
def save(r, state):
r.session.set_state(json.dumps(state))
def handle(r, msg, state, log):
step, done = msg.subject, state["done"]
if step in done: # redelivered
log.append(f"{step}: duplicate, skipped")
elif step != STEPS[len(done)]: # too early
state["deferred"][step] = msg.sequence_number
save(r, state) # remember it
r.defer_message(msg)
log.append(f"{step}: deferred")
return
else:
done.append(step) # do the work
state["deferred"].pop(step, None)
log.append(step)
save(r, state) # then settle
r.complete_message(msg)
nxt = STEPS[len(done)] if len(done) < len(STEPS) else None
if nxt in state["deferred"]: # its turn now
seq = state["deferred"][nxt]
handle(r, r.receive_deferred_messages([seq])[0], state, log)
""")
write('workflow.py', """\
import json
from azure.servicebus import NEXT_AVAILABLE_SESSION
from azure.servicebus.exceptions import OperationTimeoutError
from common import client
from steps import EMPTY, handle
with client() as sb:
while True: # one order at a time
try:
with sb.get_queue_receiver(
"order-workflow", max_wait_time=3,
session_id=NEXT_AVAILABLE_SESSION) as r:
state = json.loads(r.session.get_state() or EMPTY)
log = []
for msg in r.receive_messages(max_message_count=20,
max_wait_time=3):
handle(r, msg, state, log)
if len(state["done"]) == 4: # finished: clean up
r.session.set_state(None)
print(r.session.session_id, "|", ", ".join(log))
print(" state:", r.session.get_state())
except OperationTimeoutError:
break
""")
run('python send_workflow.py')
run('python workflow.py')
Part 7: Duplicate Detection, Scheduling and Deferral
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 Service Bus emulator (2.0.1) with SQL Server and azure-servicebus 7.15.0. Features the emulator does not support are shown as code. Part of the Real Time Azure Service Bus 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 Service Bus documentation (learn.microsoft.com/azure/service-bus-messaging) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.