In this post we are going to cover duplicate detection, scheduling and deferral 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: an order system must send each confirmation once, send payment reminders after three days, and never lose a message.
Duplicate detection drops a message the queue has already seen. Scheduling sends a message now that only shows up later. Deferral sets a message aside until you are ready for it. Interviewers ask about these three together because they all deal with timing and repeats, and because the real question behind them is whether you know that none of them makes processing exactly once.
Part 7 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: A publisher retries after a timeout and the same message arrives twice. How does duplicate detection help?
- Question 2: A teammate asks how long Service Bus remembers message IDs. How does the detection window work?
- Question 3: A reminder email must go out exactly one day after signup. How do scheduled messages work?
- Question 4: A scheduled campaign must be cancelled. How do you cancel a scheduled message?
- Question 5: A message arrives before the data it depends on. How does deferral work?
- Question 6: Someone asks why duplicate detection alone doesn't make processing exactly once. How do you answer?
- Scenario 7: An order system must send each confirmation once, send payment reminders after three days, and never lose a message. What is your design?
- The full script
The short answer
Duplicate detection drops a resend with a known MessageId inside the window, so publishers use a business key as the ID. Scheduled messages appear at their enqueue time and are cancelled by sequence number. Deferral sets a message aside until you fetch it by sequence number. None of it makes processing exactly once: consumers stay idempotent.
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 eight queues. payments, emails and confirmations have duplicate detection on, with a 20 second window, and emails has a 5 second lock. payments-nodup is the same as payments without detection, and reminders, campaigns, order-events and payment-reminders are plain queues. Two helpers come first: drain.py receives and completes everything in a queue, and peek.py peeks a queue from sequence number 1 and prints each message ID with its state.
{
"UserConfig": {
"Namespaces": [
{
"Name": "sbemulatorns",
"Queues": [
{
"Name": "payments",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": true,
"RequiresSession": false
}
},
{
"Name": "payments-nodup",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "reminders",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "campaigns",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "order-events",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "emails",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT5S",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": true,
"RequiresSession": false
}
},
{
"Name": "confirmations",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": true,
"RequiresSession": false
}
},
{
"Name": "payment-reminders",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
}
],
"Topics": []
}
],
"Logging": {
"Type": "File"
}
}
}import os, sys
from azure.servicebus import ServiceBusClient
queue = sys.argv[1]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
got = []
with client, client.get_queue_receiver(queue, max_wait_time=3) as rx:
while batch := rx.receive_messages(max_message_count=20):
for m in batch:
rx.complete_message(m)
got.append(str(m))
print(f"{queue}: received {len(got)}", *sorted(got), sep="\n ")import os, sys
from azure.servicebus import ServiceBusClient
queue = sys.argv[1]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver(queue) as rx:
peeked = rx.peek_messages(50, sequence_number=1)
print(f"{queue}: peek shows {len(peeked)}", *sorted(
f"{m.message_id} {m.state.name}" for m in peeked), sep="\n ")Question 1: A publisher retries after a timeout and the same message arrives twice. How does duplicate detection help?
Duplicate detection makes the queue remember the message ID of every message it accepted, for a time window. A send with an ID it already logged still succeeds, but the new copy is dropped. Only the message ID is compared, plus the partition key if the entity is partitioned.
Here is the flow:
- Publisher - retries after a timeout
paymentsqueue - remembers message IDs- Consumer - gets one copy
Let's try it. publish.py sends each payment twice, like a retry after a timeout. Payment 1001 has a business message ID, pay-1001. Payment 1002 has none. We run it against payments, which has detection on, and against payments-nodup, which doesn't, and drain each one: python publish.py payments, python drain.py payments, python publish.py payments-nodup and python drain.py payments-nodup.
import os, sys
from azure.servicebus import ServiceBusClient, ServiceBusMessage
queue = sys.argv[1]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender(queue) as sender:
for attempt in [1, 2]: # 2 = the retry after a timeout
sender.send_messages(ServiceBusMessage(
"charge order 1001: 49.90", message_id="pay-1001"))
sender.send_messages(ServiceBusMessage(
"charge order 1002: 15.00")) # no message_id set
print(f"{queue}: sent each payment twice")$ python publish.py payments
payments: sent each payment twice
$ python drain.py payments
payments: received 3
charge order 1001: 49.90
charge order 1002: 15.00
charge order 1002: 15.00
$ python publish.py payments-nodup
payments-nodup: sent each payment twice
$ python drain.py payments-nodup
payments-nodup: received 4
charge order 1001: 49.90
charge order 1001: 49.90
charge order 1002: 15.00
charge order 1002: 15.00With duplicate detection on, payments: received 3. Order 1001 arrives once, but order 1002 arrives twice: with no message ID of our own, there's nothing stable to compare. Without duplicate detection, payments-nodup: received 4, all four copies.
It is set when you create the queue or topic and can't be switched later. And it needs Standard or Premium, not Basic.
Follow Microsoft's advice and use a business key as the message ID, like the order number plus the action, so every retry repeats it. In Python, you set it yourself.
Question 2: A teammate asks how long Service Bus remembers message IDs. How does the detection window work?
For the duplicate detection window. It defaults to 10 minutes for queues and topics:
| Duplicate detection window | Value |
|---|---|
| Default | 10 minutes, for queues and topics |
| Minimum | 20 seconds |
| Maximum | 7 days |
| Change the window size later | Yes |
| Turn detection on or off later | No: set when the entity is created |
You can change the window size later, but not the on or off switch.
Let's test the window. Our payments queue uses 20 seconds. window.py sends pay-2001 and receives it. Then it sends it again right after the first copy was completed. Then it waits 22 seconds, past the window, and sends it once more. We run python window.py.
import os, time
from azure.servicebus import ServiceBusClient, ServiceBusMessage
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
def send_and_count(note):
with client.get_queue_sender("payments") as sender:
sender.send_messages(ServiceBusMessage(
"charge order 2001: 80.00", message_id="pay-2001"))
with client.get_queue_receiver("payments", max_wait_time=3) as rx:
got = rx.receive_messages(max_message_count=5)
for m in got:
rx.complete_message(m)
print(f"{note}: delivered {len(got)}")
with client:
send_and_count("pay-2001, first send")
send_and_count("pay-2001 again, first copy completed")
time.sleep(22) # this queue's window: 20 s
send_and_count("pay-2001 again, 22 seconds later")$ python window.py
pay-2001, first send: delivered 1
pay-2001 again, first copy completed: delivered 0
pay-2001 again, 22 seconds later: delivered 1The first send is delivered 1. The second, right after the first copy was completed, is delivered 0. The ID is still remembered, even though the message is gone. The third send, 22 seconds later, is delivered 1 again: for Service Bus, it is a new message.
So size the window to your longest retry. A publisher that resends after a long outage, beyond the window, creates real duplicates.
Don't make it huge on busy entities: Microsoft advises keeping the window as small as possible for high throughput. And scheduled messages are part of duplicate detection too.
Question 3: A reminder email must go out exactly one day after signup. How do scheduled messages work?
A scheduled message is sent now, but it only shows up in the queue at its enqueue time. In Python, you call schedule_messages with the time, or set scheduled_enqueue_time_utc on the message and send it.
The life of the message looks like this:
- Signup - send now, due in 1 day
- Scheduled - invisible to receivers
- Active - at the enqueue time
Let's schedule one. Ana signs up, and schedule.py schedules her reminder 8 seconds ahead. In production, that is one day. Then it peeks the queue, tries a receive right away, and then waits for the reminder. We run python schedule.py.
import datetime, os
from azure.servicebus import ServiceBusClient, ServiceBusMessage
signup = datetime.datetime.now(datetime.timezone.utc)
due = signup + datetime.timedelta(seconds=8) # real life: days=1
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("reminders") as sender, \
client.get_queue_receiver("reminders") as rx:
seqs = sender.schedule_messages(ServiceBusMessage(
"reminder: finish your profile, Ana", message_id="ana"), due)
print("schedule_messages returned:", seqs)
for m in rx.peek_messages(5, sequence_number=1):
print("peek:", m.message_id, m.state.name)
print("receive now:", rx.receive_messages(max_wait_time=3))
for m in rx.receive_messages(max_wait_time=15):
now = datetime.datetime.now(datetime.timezone.utc)
print("received:", str(m), "| at or after due:", now >= due)
rx.complete_message(m)$ python schedule.py
schedule_messages returned: [1]
peek: ana SCHEDULED
receive now: []
received: reminder: finish your profile, Ana | at or after due: Trueschedule_messages returns a list of sequence numbers, here just [1]. Peek shows ana SCHEDULED. The receive right away returns an empty list, []. Then the reminder arrives, and at or after due: True.
Three details matter:
- Enqueue time = available - not an exact delivery time
- No recurring schedules - for a daily job, the consumer schedules the next run
- Time to live - counts from the enqueue time, not from the send
Keep the sequence number with the signup record. You need it to cancel.
Question 4: A scheduled campaign must be cancelled. How do you cancel a scheduled message?
By its sequence number. schedule_messages returns one per message, and cancel_scheduled_messages takes them back, which deletes the messages. Setting only the scheduled time on a normal send gives you no number back.
Let's run a campaign. schedule_campaign.py schedules the spring sale for three regions, 30 minutes ahead, and saves the sequence numbers keyed by region. Then the campaign is pulled in east and west, so cancel.py cancels those two. We peek before and after: python schedule_campaign.py, python peek.py campaigns, python cancel.py east west and python peek.py campaigns.
import datetime, json, os
from azure.servicebus import ServiceBusClient, ServiceBusMessage
regions = ["east", "west", "north"]
due = datetime.datetime.now(datetime.timezone.utc) + \
datetime.timedelta(minutes=30) # Friday 9 AM in real life
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("campaigns") as sender:
msgs = [ServiceBusMessage(f"spring sale: {r}",
message_id=f"spring-{r}")
for r in regions]
seqs = sender.schedule_messages(msgs, due)
saved = dict(zip(regions, seqs)) # keep them: needed to cancel
json.dump(saved, open("campaign_seqs.json", "w"))
print("scheduled, sequence numbers:", saved)import json, os, sys
from azure.servicebus import ServiceBusClient
saved = json.load(open("campaign_seqs.json"))
seqs = [saved[region] for region in sys.argv[1:]]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("campaigns") as sender:
sender.cancel_scheduled_messages(seqs)
print("cancelled sequence numbers", seqs, "for", sys.argv[1:])$ python schedule_campaign.py
scheduled, sequence numbers: {'east': 1, 'west': 2, 'north': 3}
$ python peek.py campaigns
campaigns: peek shows 3
spring-east SCHEDULED
spring-north SCHEDULED
spring-west SCHEDULED
$ python cancel.py east west
cancelled sequence numbers [1, 2] for ['east', 'west']
$ python peek.py campaigns
campaigns: peek shows 1
spring-north SCHEDULEDThe saved numbers are {'east': 1, 'west': 2, 'north': 3}, and peek shows three SCHEDULED messages. We cancel [1, 2] for east and west, and peek now shows only spring-north SCHEDULED.
There are two catches. The number is only valid while the message is scheduled: on activation, it gets a new one. And a cancel near the due time races with activation, so cancel early, and let the consumer check that the campaign is still on.
Store the sequence numbers in the campaign table, not in memory. A restarted service that lost them has to peek the queue to find them again.
Question 5: A message arrives before the data it depends on. How does deferral work?
Deferral sets a message aside without losing it. The consumer calls defer, the message stays in the queue, but normal receives skip it. You get it back only by its sequence number, with receive_deferred_messages.
Let's create the problem. The payment for order 2001 arrives before the order placed event. The consumer doesn't know the order yet, so it saves the sequence number in its own database, then defers the payment. Later the order placed event arrives. We run python send_event.py payment 2001, python consumer.py, python peek.py order-events, python send_event.py placed 2001, python consumer.py and python peek.py order-events.
import os, sys
from azure.servicebus import ServiceBusClient, ServiceBusMessage
kind, order = sys.argv[1], sys.argv[2]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("order-events") as sender:
sender.send_messages(ServiceBusMessage(
order, subject=kind, message_id=f"{order}-{kind}"))
print("sent", kind, "for order", order)import json, os
from azure.servicebus import ServiceBusClient
db = json.load(open("db.json")) if os.path.exists("db.json") \
else {"orders": [], "deferred": {}} # our own database
def save(): # write it to disk now
json.dump(db, open("db.json", "w"))
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver("order-events",
max_wait_time=3) as rx:
for m in rx.receive_messages(max_message_count=10):
kind, order = m.subject, str(m)
if kind == "payment" and order not in db["orders"]:
db["deferred"][order] = m.sequence_number
save() # keep the number, THEN defer
rx.defer_message(m) # set aside, not lost
print(f"payment {order}: order unknown, deferred,"
f" sequence number {m.sequence_number}")
continue
db["orders"].append(order) # process: save the order
save() # saved, then complete
rx.complete_message(m)
print(f"{kind} {order}: processed")
if order in db["deferred"]: # its payment is waiting
seq = db["deferred"][order]
for d in rx.receive_deferred_messages([seq]):
rx.complete_message(d)
print(f"payment {order}: deferred {seq} processed")
db["deferred"].pop(order) # done: forget the number
save()$ python send_event.py payment 2001
sent payment for order 2001
$ python consumer.py
payment 2001: order unknown, deferred, sequence number 1
$ python peek.py order-events
order-events: peek shows 1
2001-payment DEFERRED
$ python send_event.py placed 2001
sent placed for order 2001
$ python consumer.py
placed 2001: processed
payment 2001: deferred 1 processed
$ python peek.py order-events
order-events: peek shows 0The first consumer run prints order unknown, deferred, sequence number 1, and peek shows 2001-payment DEFERRED. After the order placed event, the consumer saves the order, finds the waiting payment, receives it by sequence number 1, and completes it: payment 2001: deferred 1 processed. Peek then shows 0.
Microsoft uses this very example, a payment before the purchase order. Deferral beats abandon here. Abandon puts it straight back, it fails again, and it burns attempts toward the max delivery count.
Lose the sequence number and the message just sits there. Deferred messages don't expire into the dead-letter queue on their own, so save the number durably before you defer, and sweep for old ones with peek.
Question 6: Someone asks why duplicate detection alone doesn't make processing exactly once. How do you answer?
Because it only works on the send side: it drops a second send with the same message ID inside the window. But delivery is at least once. If a lock expires, or the consumer crashes before completing, the same message comes again, and that is not a new send.
Let's prove it. send_job.py sends the email job email-3001 twice, and duplicate detection keeps one. The worker's job is a database insert: an email row, with the message ID, in one commit. In slow mode the work takes 7 seconds, and the emails queue's lock is 5. Then we run the worker again. The commands are python send_job.py, python worker.py slow and python worker.py.
import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("emails") as sender:
for attempt in [1, 2]: # a publisher retry
sender.send_messages(ServiceBusMessage(
"order 3001 confirmation email", message_id="email-3001"))
print("sent email-3001 twice")import os, sqlite3, sys, time
from azure.servicebus import ServiceBusClient
from azure.servicebus.exceptions import MessageLockLostError
db = sqlite3.connect("emails.db")
db.execute("CREATE TABLE IF NOT EXISTS outbox" # a mailer sends
" (message_id TEXT PRIMARY KEY, body TEXT)") # its rows
ADD = "INSERT OR IGNORE INTO outbox VALUES (?, ?)" # skip if seen
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver("emails",
max_wait_time=3) as rx:
for m in rx.receive_messages(max_message_count=5):
print(m.message_id, "delivery_count", m.delivery_count)
with db: # the work and its message_id, one commit
new = db.execute(ADD, (m.message_id, str(m))).rowcount
print(" email row saved" if new else " already saved: skip")
if sys.argv[1:] == ["slow"]:
time.sleep(7) # longer than the 5 s lock
try:
rx.complete_message(m)
print(" completed")
except MessageLockLostError as e:
print(" complete failed:", type(e).__name__, "|", e)
count = db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
print("email rows for the mailer:", count)$ python send_job.py
sent email-3001 twice
$ python worker.py slow
email-3001 delivery_count 0
email row saved
complete failed: MessageLockLostError | The lock on the message lock has expired.
email rows for the mailer: 1
$ python worker.py
email-3001 delivery_count 1
already saved: skip
completed
email rows for the mailer: 1The worker gets it with delivery_count 0, zero on the first try, and saves the email row. Then complete fails with MessageLockLostError: the lock has expired. The message comes back with delivery_count 1. The second run prints already saved: skip and completes it, and the table holds email rows for the mailer: 1, not two.
A real email call can't join that commit, so a mailer sends from the table, with an idempotency key if the provider has one.
| Duplicate detection | Idempotent consumer | |
|---|---|---|
| Where | the broker, on send | your consumer and database |
| Catches | the same MessageId sent twice in the window | the same message delivered twice |
| Misses | redelivery after a lock expiry or crash | nothing, if check and DB work share a commit |
Duplicate detection plus an idempotent consumer. Microsoft says exactly once processing needs extra logic in your application, usually a check on the message ID or a business key.
Scenario 7: An order system must send each confirmation once, send payment reminders after three days, and never lose a message. What is your design?
Three features together. Confirmations go to a queue with duplicate detection, and the message ID is confirm- plus the order number. Payment reminders are scheduled messages, three days ahead, cancelled when the customer pays.
Here is the design:
- Order service - confirm + schedule
confirmations- duplicate detectionpayment-reminders- scheduled, 3 days- Email consumer - idempotent
- Payment received - cancel by sequence number
Let's run it. place_orders.py places two orders. Each confirmation is sent twice, a retry, and each reminder is scheduled three days out, with its sequence number saved in the orders table. Then order 4001 is paid. We run python place_orders.py, python pay.py 4001, python drain.py confirmations and python peek.py payment-reminders.
import datetime, os, sqlite3
from azure.servicebus import ServiceBusClient, ServiceBusMessage
db = sqlite3.connect("shop.db")
db.execute("CREATE TABLE IF NOT EXISTS orders"
" (order_id TEXT PRIMARY KEY, reminder_seq INT)")
in_3_days = datetime.datetime.now(datetime.timezone.utc) + \
datetime.timedelta(days=3)
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("confirmations") as conf, \
client.get_queue_sender("payment-reminders") as remind:
for order in ["4001", "4002"]:
# a retry sends the confirmation twice, same message_id
body, mid = f"order {order} confirmed", f"confirm-{order}"
conf.send_messages(ServiceBusMessage(body, message_id=mid))
conf.send_messages(ServiceBusMessage(body, message_id=mid))
[seq] = remind.schedule_messages(ServiceBusMessage(
f"pay order {order}", message_id=f"remind-{order}"),
in_3_days)
db.execute("INSERT INTO orders VALUES (?, ?)", (order, seq))
db.commit()
print(f"order {order}: confirmed, reminder sequence {seq}")import os, sqlite3, sys
from azure.servicebus import ServiceBusClient
order = sys.argv[1]
db = sqlite3.connect("shop.db")
[seq] = db.execute("SELECT reminder_seq FROM orders"
" WHERE order_id = ?", (order,)).fetchone()
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("payment-reminders") as remind:
remind.cancel_scheduled_messages(seq)
print(f"order {order} paid: reminder sequence {seq} cancelled")$ python place_orders.py
order 4001: confirmed, reminder sequence 1
order 4002: confirmed, reminder sequence 2
$ python pay.py 4001
order 4001 paid: reminder sequence 1 cancelled
$ python drain.py confirmations
confirmations: received 2
order 4001 confirmed
order 4002 confirmed
$ python peek.py payment-reminders
payment-reminders: peek shows 1
remind-4002 SCHEDULEDThe reminders get sequence numbers 1 and 2. Order 4001 is paid, so reminder sequence 1 cancelled. The confirmations queue delivers 2 messages, one per order. Peek shows only remind-4002 SCHEDULED.
Never lose a message means three more things:
- Peek-lock - complete after the work
- Max delivery count + DLQ - with an alert
- Idempotent consumers - redelivery still happens
One honest gap: the demo writes the database after the sends, so a crash in between leaves a reminder we can't cancel. The reminder consumer checks that the order is still unpaid.
A transactional outbox closes that gap. That is the next part of this series.
Key points to remember
- Duplicate detection drops a resend with a known MessageId: use a business key
- The window: 10 minutes by default, 20 seconds to 7 days; enable it at creation
- schedule_messages returns sequence numbers; cancel with them while still scheduled
- Defer a message that came too early; fetch it later by sequence number
- Not exactly once: redelivery still happens, so consumers stay idempotent
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 #7 - Duplicate detection, scheduling and deferral
# 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_07_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": "payments",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": true,
"RequiresSession": false
}
},
{
"Name": "payments-nodup",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "reminders",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "campaigns",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "order-events",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
},
{
"Name": "emails",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT5S",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": true,
"RequiresSession": false
}
},
{
"Name": "confirmations",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": true,
"RequiresSession": false
}
},
{
"Name": "payment-reminders",
"Properties": {
"DeadLetteringOnMessageExpiration": false,
"DefaultMessageTimeToLive": "PT1H",
"DuplicateDetectionHistoryTimeWindow": "PT20S",
"ForwardDeadLetteredMessagesTo": "",
"ForwardTo": "",
"LockDuration": "PT1M",
"MaxDeliveryCount": 3,
"RequiresDuplicateDetection": false,
"RequiresSession": false
}
}
],
"Topics": []
}
],
"Logging": {
"Type": "File"
}
}
}
""")
# Step 0: the starting files
write('drain.py', 'import os, sys\nfrom azure.servicebus import ServiceBusClient\n\nqueue = sys.argv[1]\nclient = ServiceBusClient.from_connection_string(\n os.environ["SERVICEBUS_CONNECTION_STRING"])\ngot = []\nwith client, client.get_queue_receiver(queue, max_wait_time=3) as rx:\n while batch := rx.receive_messages(max_message_count=20):\n for m in batch:\n rx.complete_message(m)\n got.append(str(m))\nprint(f"{queue}: received {len(got)}", *sorted(got), sep="\\n ")\n')
write('peek.py', 'import os, sys\nfrom azure.servicebus import ServiceBusClient\n\nqueue = sys.argv[1]\nclient = ServiceBusClient.from_connection_string(\n os.environ["SERVICEBUS_CONNECTION_STRING"])\nwith client, client.get_queue_receiver(queue) as rx:\n peeked = rx.peek_messages(50, sequence_number=1)\nprint(f"{queue}: peek shows {len(peeked)}", *sorted(\n f"{m.message_id} {m.state.name}" for m in peeked), sep="\\n ")\n')
# Step 1: A publisher retry with and without a stable message_id, with and without duplicate detection
write('publish.py', """\
import os, sys
from azure.servicebus import ServiceBusClient, ServiceBusMessage
queue = sys.argv[1]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender(queue) as sender:
for attempt in [1, 2]: # 2 = the retry after a timeout
sender.send_messages(ServiceBusMessage(
"charge order 1001: 49.90", message_id="pay-1001"))
sender.send_messages(ServiceBusMessage(
"charge order 1002: 15.00")) # no message_id set
print(f"{queue}: sent each payment twice")
""")
run('python publish.py payments')
run('python drain.py payments')
run('python publish.py payments-nodup')
run('python drain.py payments-nodup')
# Step 2: The duplicate detection window: a completed message is still remembered, 22 seconds later is new
write('window.py', """\
import os, time
from azure.servicebus import ServiceBusClient, ServiceBusMessage
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
def send_and_count(note):
with client.get_queue_sender("payments") as sender:
sender.send_messages(ServiceBusMessage(
"charge order 2001: 80.00", message_id="pay-2001"))
with client.get_queue_receiver("payments", max_wait_time=3) as rx:
got = rx.receive_messages(max_message_count=5)
for m in got:
rx.complete_message(m)
print(f"{note}: delivered {len(got)}")
with client:
send_and_count("pay-2001, first send")
send_and_count("pay-2001 again, first copy completed")
time.sleep(22) # this queue's window: 20 s
send_and_count("pay-2001 again, 22 seconds later")
""")
run('python window.py')
# Step 3: schedule_messages: a reminder that appears only at its time
write('schedule.py', 'import datetime, os\nfrom azure.servicebus import ServiceBusClient, ServiceBusMessage\n\nsignup = datetime.datetime.now(datetime.timezone.utc)\ndue = signup + datetime.timedelta(seconds=8) # real life: days=1\nclient = ServiceBusClient.from_connection_string(\n os.environ["SERVICEBUS_CONNECTION_STRING"])\nwith client, client.get_queue_sender("reminders") as sender, \\\n client.get_queue_receiver("reminders") as rx:\n seqs = sender.schedule_messages(ServiceBusMessage(\n "reminder: finish your profile, Ana", message_id="ana"), due)\n print("schedule_messages returned:", seqs)\n for m in rx.peek_messages(5, sequence_number=1):\n print("peek:", m.message_id, m.state.name)\n print("receive now:", rx.receive_messages(max_wait_time=3))\n for m in rx.receive_messages(max_wait_time=15):\n now = datetime.datetime.now(datetime.timezone.utc)\n print("received:", str(m), "| at or after due:", now >= due)\n rx.complete_message(m)\n')
run('python schedule.py')
# Step 4: Cancel scheduled messages by sequence number (scheduled 30 minutes ahead)
write('schedule_campaign.py', 'import datetime, json, os\nfrom azure.servicebus import ServiceBusClient, ServiceBusMessage\n\nregions = ["east", "west", "north"]\ndue = datetime.datetime.now(datetime.timezone.utc) + \\\n datetime.timedelta(minutes=30) # Friday 9 AM in real life\nclient = ServiceBusClient.from_connection_string(\n os.environ["SERVICEBUS_CONNECTION_STRING"])\nwith client, client.get_queue_sender("campaigns") as sender:\n msgs = [ServiceBusMessage(f"spring sale: {r}",\n message_id=f"spring-{r}")\n for r in regions]\n seqs = sender.schedule_messages(msgs, due)\nsaved = dict(zip(regions, seqs)) # keep them: needed to cancel\njson.dump(saved, open("campaign_seqs.json", "w"))\nprint("scheduled, sequence numbers:", saved)\n')
write('cancel.py', """\
import json, os, sys
from azure.servicebus import ServiceBusClient
saved = json.load(open("campaign_seqs.json"))
seqs = [saved[region] for region in sys.argv[1:]]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("campaigns") as sender:
sender.cancel_scheduled_messages(seqs)
print("cancelled sequence numbers", seqs, "for", sys.argv[1:])
""")
run('python schedule_campaign.py')
run('python peek.py campaigns')
run('python cancel.py east west')
run('python peek.py campaigns')
# Step 5: Deferral: a payment that arrives before its order
write('send_event.py', """\
import os, sys
from azure.servicebus import ServiceBusClient, ServiceBusMessage
kind, order = sys.argv[1], sys.argv[2]
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("order-events") as sender:
sender.send_messages(ServiceBusMessage(
order, subject=kind, message_id=f"{order}-{kind}"))
print("sent", kind, "for order", order)
""")
write('consumer.py', 'import json, os\nfrom azure.servicebus import ServiceBusClient\n\ndb = json.load(open("db.json")) if os.path.exists("db.json") \\\n else {"orders": [], "deferred": {}} # our own database\ndef save(): # write it to disk now\n json.dump(db, open("db.json", "w"))\nclient = ServiceBusClient.from_connection_string(\n os.environ["SERVICEBUS_CONNECTION_STRING"])\nwith client, client.get_queue_receiver("order-events",\n max_wait_time=3) as rx:\n for m in rx.receive_messages(max_message_count=10):\n kind, order = m.subject, str(m)\n if kind == "payment" and order not in db["orders"]:\n db["deferred"][order] = m.sequence_number\n save() # keep the number, THEN defer\n rx.defer_message(m) # set aside, not lost\n print(f"payment {order}: order unknown, deferred,"\n f" sequence number {m.sequence_number}")\n continue\n db["orders"].append(order) # process: save the order\n save() # saved, then complete\n rx.complete_message(m)\n print(f"{kind} {order}: processed")\n if order in db["deferred"]: # its payment is waiting\n seq = db["deferred"][order]\n for d in rx.receive_deferred_messages([seq]):\n rx.complete_message(d)\n print(f"payment {order}: deferred {seq} processed")\n db["deferred"].pop(order) # done: forget the number\n save()\n')
run('python send_event.py payment 2001')
run('python consumer.py')
run('python peek.py order-events')
run('python send_event.py placed 2001')
run('python consumer.py')
run('python peek.py order-events')
# Step 6: Duplicate detection drops the resend, but a lock expiry still redelivers: an idempotent consumer
write('send_job.py', """\
import os
from azure.servicebus import ServiceBusClient, ServiceBusMessage
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("emails") as sender:
for attempt in [1, 2]: # a publisher retry
sender.send_messages(ServiceBusMessage(
"order 3001 confirmation email", message_id="email-3001"))
print("sent email-3001 twice")
""")
write('worker.py', """\
import os, sqlite3, sys, time
from azure.servicebus import ServiceBusClient
from azure.servicebus.exceptions import MessageLockLostError
db = sqlite3.connect("emails.db")
db.execute("CREATE TABLE IF NOT EXISTS outbox" # a mailer sends
" (message_id TEXT PRIMARY KEY, body TEXT)") # its rows
ADD = "INSERT OR IGNORE INTO outbox VALUES (?, ?)" # skip if seen
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_receiver("emails",
max_wait_time=3) as rx:
for m in rx.receive_messages(max_message_count=5):
print(m.message_id, "delivery_count", m.delivery_count)
with db: # the work and its message_id, one commit
new = db.execute(ADD, (m.message_id, str(m))).rowcount
print(" email row saved" if new else " already saved: skip")
if sys.argv[1:] == ["slow"]:
time.sleep(7) # longer than the 5 s lock
try:
rx.complete_message(m)
print(" completed")
except MessageLockLostError as e:
print(" complete failed:", type(e).__name__, "|", e)
count = db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0]
print("email rows for the mailer:", count)
""")
run('python send_job.py')
run('python worker.py slow')
run('python worker.py')
# Step 7: Confirmations once, reminders in three days, cancelled when paid
write('place_orders.py', 'import datetime, os, sqlite3\nfrom azure.servicebus import ServiceBusClient, ServiceBusMessage\n\ndb = sqlite3.connect("shop.db")\ndb.execute("CREATE TABLE IF NOT EXISTS orders"\n " (order_id TEXT PRIMARY KEY, reminder_seq INT)")\nin_3_days = datetime.datetime.now(datetime.timezone.utc) + \\\n datetime.timedelta(days=3)\nclient = ServiceBusClient.from_connection_string(\n os.environ["SERVICEBUS_CONNECTION_STRING"])\nwith client, client.get_queue_sender("confirmations") as conf, \\\n client.get_queue_sender("payment-reminders") as remind:\n for order in ["4001", "4002"]:\n # a retry sends the confirmation twice, same message_id\n body, mid = f"order {order} confirmed", f"confirm-{order}"\n conf.send_messages(ServiceBusMessage(body, message_id=mid))\n conf.send_messages(ServiceBusMessage(body, message_id=mid))\n [seq] = remind.schedule_messages(ServiceBusMessage(\n f"pay order {order}", message_id=f"remind-{order}"),\n in_3_days)\n db.execute("INSERT INTO orders VALUES (?, ?)", (order, seq))\n db.commit()\n print(f"order {order}: confirmed, reminder sequence {seq}")\n')
write('pay.py', """\
import os, sqlite3, sys
from azure.servicebus import ServiceBusClient
order = sys.argv[1]
db = sqlite3.connect("shop.db")
[seq] = db.execute("SELECT reminder_seq FROM orders"
" WHERE order_id = ?", (order,)).fetchone()
client = ServiceBusClient.from_connection_string(
os.environ["SERVICEBUS_CONNECTION_STRING"])
with client, client.get_queue_sender("payment-reminders") as remind:
remind.cancel_scheduled_messages(seq)
print(f"order {order} paid: reminder sequence {seq} cancelled")
""")
run('python place_orders.py')
run('python pay.py 4001')
run('python drain.py confirmations')
run('python peek.py payment-reminders')
Part 8: Transactions and Reliable Messaging Patterns
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.