In this post we are going to cover acknowledgements and redelivery in Pub/Sub. These are 7 real Pub/Sub interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: a payments consumer writes to a database and must never double-charge a customer, and you design the acks and the writes.
An ack is how a consumer tells Pub/Sub "I'm done with this message". If the ack doesn't arrive in time, the message comes back. Interviewers love this topic because it is where real bugs live: the same order processed twice, an order lost because it was acked too early, a message stuck in a nack loop. If you can explain when a message comes back and how to make that harmless, you show that you have run consumers in production.
Part 4 of the Real Time Pub/Sub Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: The same order was processed twice. Why does Pub/Sub deliver a message more than once?
- Question 2: A consumer crashed in the middle of processing. What happens to its unacked messages?
- Question 3: A teammate acks the message before writing to the database. What is the risk?
- Question 4: Some messages take ten minutes to process. How do you extend the ack deadline?
- Question 5: A consumer wants to give a message back right away. What does a nack do?
- Question 6: Your manager asks how to make duplicate deliveries harmless. How do you design idempotent consumers?
- Scenario 7: A payments consumer writes to a database and must never double-charge a customer. How do you design acks and writes?
- The full script
The short answer
Pub/Sub delivery is at least once: a message that isn't acked before its ack deadline comes back, and duplicates also come from lost acks and publisher retries. Ack only after the commit, let the client extend deadlines for long work, nack for transient problems, and make every consumer idempotent with a business key.
Let's set up the project
We run Google's official Pub/Sub emulator locally, with small Python scripts that use the google-cloud-pubsub client, so every output is real. Features the emulator doesn't support are shown as code. Before the first step we run python setup.py once. It creates seven topics in the project tb-shop-dev, each with its own pull subscriptions, and a small SQLite database, shop.db, from schema.sql. db.py saves an order, and DB_DOWN=1 simulates a database outage.
CREATE TABLE orders (order_id INTEGER PRIMARY KEY);
CREATE TABLE credits (order_id INTEGER PRIMARY KEY);
-- payment_id is the idempotency key
CREATE TABLE charges (payment_id TEXT PRIMARY KEY,
amount NUMERIC NOT NULL);import sqlite3
from google.cloud import pubsub_v1
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topics = {"orders": ["fulfil-pull"], "shipments": ["packing-pull"],
"ledger": ["ledger-pull"], "emails": ["email-pull"],
"reports": ["reports-stream", "reports-pull"],
"credits": ["credits-pull"], "payments": []}
for topic_id, subs in topics.items():
topic = pub.topic_path("tb-shop-dev", topic_id)
pub.create_topic(name=topic)
for sub_id in subs:
path = sub.subscription_path("tb-shop-dev", sub_id)
sub.create_subscription(name=path, topic=topic)
with sqlite3.connect("shop.db") as db:
db.executescript(open("schema.sql").read())import json, sys
from google.cloud import pubsub_v1
topic_id = sys.argv[1]
count, start = int(sys.argv[2]), int(sys.argv[3])
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", topic_id)
for order_id in range(start, start + count):
data = json.dumps({"order_id": order_id}).encode()
message_id = pub.publish(topic, data).result() # wait for each
print("published order", order_id, "as message", message_id)import os, sqlite3
def save_order(order_id):
"""Write one order. DB_DOWN=1 simulates a database outage."""
if os.environ.get("DB_DOWN"):
raise ConnectionError("database is down")
with sqlite3.connect("shop.db") as db:
db.execute("INSERT INTO orders VALUES (?)", (order_id,))
if __name__ == "__main__":
with sqlite3.connect("shop.db") as db:
rows = db.execute("SELECT order_id FROM orders ORDER BY 1")
print("orders in the database:", [r[0] for r in rows])Question 1: The same order was processed twice. Why does Pub/Sub deliver a message more than once?
Because delivery is at least once. Pub/Sub would rather deliver twice than lose a message, so a message can come back, occasionally even after it was acked.
The most common cause is a slow consumer. Let's publish order 1001 to orders and run python slow_worker.py. Worker A pulls it and works for 13 seconds on a 10 second ack deadline. Then worker B pulls.
import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "fulfil-pull")
resp = sub.pull(subscription=path, max_messages=1) # worker A
a = resp.received_messages[0].message
print("worker A got message", a.message_id, a.data.decode())
time.sleep(13) # slow work: past the 10 s deadline
resp = sub.pull(subscription=path, max_messages=1) # worker B
rb = resp.received_messages[0]
b = rb.message
print("worker B got message", b.message_id, b.data.decode())
sub.acknowledge(subscription=path, ack_ids=[rb.ack_id])
print("same order processed twice:", a.data == b.data)$ python publish.py orders 1 1001
published order 1001 as message 1
$ python slow_worker.py
worker A got message 1 {"order_id": 1001}
worker B got message 1 {"order_id": 1001}
same order processed twice: Trueworker A got message 1 {"order_id": 1001}, and then worker B got message 1 {"order_id": 1001}. The deadline expired, worker B got the same message, and same order processed twice: True.
Slow work is only one of the causes:
- Slow work - the ack deadline expires
- Lost ack - on the way back
- Overloaded subscriber - acks too late
- Publisher retry - same data, new message ID
So the real interview answer isn't how to stop duplicates. It's how to make them harmless, with idempotent consumers, which we build later in this post.
I log the message ID and the business key on every delivery. Then a double is easy to explain: a redelivery keeps the message ID, and a publisher retry gets a new one.
Question 2: A consumer crashed in the middle of processing. What happens to its unacked messages?
Nothing is lost. Unacked messages stay in the subscription, and once their ack deadline expires, Pub/Sub delivers them again, to the same or another subscriber.
Let's publish three orders to shipments, 2001 to 2003. Then python crash.py pulls all three, acks the first, and crashes on the second. Then python restart.py starts a new worker.
import os
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "packing-pull")
got = []
while len(got) < 3:
resp = sub.pull(subscription=path, max_messages=3)
got += resp.received_messages
for r in sorted(got, key=lambda r: r.message.data):
print("packing", r.message.data.decode())
if b"2002" in r.message.data:
print("crash! no ack for this order or the next")
os._exit(1) # the pod is killed
sub.acknowledge(subscription=path, ack_ids=[r.ack_id])
print(" acked")import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "packing-pull")
resp = sub.pull(subscription=path, max_messages=10, timeout=6)
n = len(resp.received_messages)
print("new worker, right away:", n, "messages")
got, start = [], time.time()
while len(got) < 2 and time.time() - start < 30:
resp = sub.pull(subscription=path, max_messages=10)
got += resp.received_messages
orders = sorted(r.message.data.decode() for r in got)
print("after the ack deadline:", orders)
sub.acknowledge(subscription=path, ack_ids=[r.ack_id for r in got])$ python publish.py shipments 3 2001
published order 2001 as message 2
published order 2002 as message 3
published order 2003 as message 4
$ python crash.py
packing {"order_id": 2001}
acked
packing {"order_id": 2002}
crash! no ack for this order or the next
$ python restart.py
new worker, right away: 0 messages
after the ack deadline: ['{"order_id": 2002}', '{"order_id": 2003}']The first worker prints packing {"order_id": 2001} and acked, then packing {"order_id": 2002} and crash! no ack for this order or the next.
The new worker starts right away and gets 0 messages, because the crashed worker's leases are still running. After the ack deadline, both unacked orders come back: ['{"order_id": 2002}', '{"order_id": 2003}']. Order 2001 was acked, so it's done.
With streaming pull, the client keeps extending leases while it runs. When the process dies, the extensions stop, and messages come back when the current lease ends. Setting max_duration_per_lease_extension (off by default) caps that wait.
Make processing restartable, because the half-done order runs again. And shut down cleanly: cancel the stream, and the Python client nacks what it hasn't started, so other workers get it right away.
Question 3: A teammate acks the message before writing to the database. What is the risk?
Losing data. Once a message is acked, Pub/Sub discards it by default. If the write then fails, nothing redelivers it.
Let's prove it. We publish order 4001 to ledger, and run ack_first.py with the database down, then again with it up. Then python db.py lists what's in the database.
import json
from google.cloud import pubsub_v1
from db import save_order
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "ledger-pull")
resp = sub.pull(subscription=path, max_messages=10, timeout=10)
print("messages pulled:", len(resp.received_messages))
for r in resp.received_messages:
order_id = json.loads(r.message.data)["order_id"]
sub.acknowledge(subscription=path, ack_ids=[r.ack_id]) # too soon
print("acked order", order_id)
try:
save_order(order_id)
print("saved order", order_id)
except ConnectionError as e:
print("save failed:", e)$ python publish.py ledger 1 4001
published order 4001 as message 5
$ DB_DOWN=1 python ack_first.py
messages pulled: 1
acked order 4001
save failed: database is down
$ python ack_first.py
messages pulled: 0
$ python db.py
orders in the database: []With DB_DOWN=1, we get acked order 4001, then save failed: database is down. The second run says messages pulled: 0. And orders in the database: []. The order is gone.
Now write first, ack after. We publish order 4002 and run ack_after.py the same way.
import json
from google.cloud import pubsub_v1
from db import save_order
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "ledger-pull")
resp = sub.pull(subscription=path, max_messages=10, timeout=10)
print("messages pulled:", len(resp.received_messages))
for r in resp.received_messages:
order_id = json.loads(r.message.data)["order_id"]
try:
save_order(order_id) # write first
sub.acknowledge(subscription=path, ack_ids=[r.ack_id])
print("saved, then acked order", order_id)
except ConnectionError as e:
sub.modify_ack_deadline(subscription=path, ack_ids=[r.ack_id],
ack_deadline_seconds=0) # a nack
print("save failed:", e, "- nacked order", order_id)$ python publish.py ledger 1 4002
published order 4002 as message 6
$ DB_DOWN=1 python ack_after.py
messages pulled: 1
save failed: database is down - nacked order 4002
$ python ack_after.py
messages pulled: 1
saved, then acked order 4002
$ python db.py
orders in the database: [4002]With the database down: save failed: database is down - nacked order 4002. With the database back, it's delivered again: saved, then acked order 4002. And the database now has [4002].
The other direction has a cost too. If the worker dies after the write but before the ack, the order comes back and is written again. That's why ack after the write always goes with an idempotent write.
Process, commit, then ack. Never ack on receipt just to keep the backlog small.
Question 4: Some messages take ten minutes to process. How do you extend the ack deadline?
With the client library, we mostly don't have to. While our callback runs, it keeps extending the deadline for us, up to max lease duration: one hour by default in the Python client.
Let's publish report 8001 to reports and run python long_job.py. The callback works for 25 seconds on a subscription with a 10 second deadline.
import time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import FlowControl
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "reports-stream")
deliveries = []
def callback(message):
deliveries.append(message.message_id)
time.sleep(25) # 25 s of work, 10 s ack deadline
message.ack()
flow = FlowControl(max_messages=1, # slow work: hold few
max_lease_duration=3600) # the default: 1 hour
future = sub.subscribe(path, callback=callback, flow_control=flow)
try:
future.result(timeout=30)
except TimeoutError:
future.cancel()
future.result()
print("ack deadline 10 s, work 25 s, deliveries:", len(deliveries))$ python publish.py reports 1 8001
published order 8001 as message 7
$ python long_job.py
ack deadline 10 s, work 25 s, deliveries: 1ack deadline 10 s, work 25 s, deliveries: 1. One delivery: lease management kept it alive. For slow work, keep max messages low, so the client doesn't hold messages it can't start for a long time.
With unary pull, we extend it ourselves with modify_ack_deadline. The same report sits on a second subscription, reports-pull. python extend.py extends it to 60 seconds, then pulls again 15 seconds later.
import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "reports-pull")
r = sub.pull(subscription=path, max_messages=1).received_messages[0]
sub.modify_ack_deadline(subscription=path, ack_ids=[r.ack_id],
ack_deadline_seconds=60) # 600 s at most
print("pulled message", r.message.message_id, "- deadline now 60 s")
time.sleep(15)
resp = sub.pull(subscription=path, max_messages=1, timeout=6)
print("pull 15 s later:", len(resp.received_messages), "messages")
sub.acknowledge(subscription=path, ack_ids=[r.ack_id])
print("acked when the work was done")$ python extend.py
pulled message 7 - deadline now 60 s
pull 15 s later: 0 messages
acked when the work was donepulled message 7 - deadline now 60 s, and pull 15 s later: 0 messages. It's still ours, and we ack it when the work is done.
One call can set up to 600 seconds. So for ten minutes, extend in a loop, like a heartbeat, before the deadline runs out.
If work can run longer than an hour, raise max lease duration. Or better, record the job somewhere durable, ack the message, and let a workflow run the long part.
Question 5: A consumer wants to give a message back right away. What does a nack do?
A nack tells Pub/Sub: I won't process this now. It's the same as setting the ack deadline to zero, so the message is ready for redelivery right away, to this subscriber or another one.
Let's publish one email event and run python nack.py. The callback nacks the first two deliveries and acks the third.
import collections
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "email-pull")
seen = collections.Counter()
def callback(message):
mid = message.message_id
seen[mid] += 1
if seen[mid] < 3:
print("delivery", seen[mid], "of message", mid, "- nack")
message.nack() # give it back now
else:
print("delivery", seen[mid], "of message", mid, "- ack")
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=5)
except TimeoutError:
future.cancel()
future.result()$ python publish.py emails 1 9001
published order 9001 as message 8
$ python nack.py
delivery 1 of message 8 - nack
delivery 2 of message 8 - nack
delivery 3 of message 8 - ackdelivery 1 of message 8 - nack, delivery 2 of message 8 - nack, then delivery 3 of message 8 - ack. Three deliveries, back to back, because the subscription has no retry policy.
That's the danger: a message that always fails becomes a hot loop. Add a retry policy for exponential backoff, 10 to 600 seconds by default once it's set, and a dead-letter topic. Every nack counts as one delivery attempt.
gcloud pubsub subscriptions update email-pull \
--min-retry-delay=10s --max-retry-delay=600s \
--dead-letter-topic=emails-dlq \
--max-delivery-attempts=5Good reasons to nack:
- Shutting down - another worker takes it now
- Dependency down for a moment - with retry backoff
A bad reason is bad data. A message that can never succeed belongs in a dead-letter topic, not in a nack loop.
A nack is modify_ack_deadline(ack_deadline_seconds=0), exactly what our ack after write example did.
Question 6: Your manager asks how to make duplicate deliveries harmless. How do you design idempotent consumers?
Idempotent means processing the same event twice has the same effect as processing it once. The key is a business key from the event, like the order ID, and a write that checks it.
Let's simulate a publisher retry. Order 6001 is published twice, then order 6002. Then python idempotent.py inserts into the credits table, where order_id is the primary key, with ON CONFLICT (order_id) DO NOTHING.
import json, sqlite3
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "credits-pull")
db, results = sqlite3.connect("shop.db"), []
while True:
resp = sub.pull(subscription=path, max_messages=10, timeout=10)
if not resp.received_messages:
break
for r in resp.received_messages:
order_id = json.loads(r.message.data)["order_id"]
with db: # one transaction
cur = db.execute("INSERT INTO credits VALUES (?)"
" ON CONFLICT (order_id) DO NOTHING",
(order_id,))
done = "saved" if cur.rowcount else "duplicate, skipped"
results.append(f"order {order_id}: {done}")
sub.acknowledge(subscription=path, ack_ids=[r.ack_id])
print("\n".join(sorted(results)))
rows = db.execute("SELECT COUNT(*) FROM credits").fetchone()[0]
print("rows in credits:", rows)$ python publish.py credits 1 6001
published order 6001 as message 9
$ python publish.py credits 1 6001
published order 6001 as message 10
$ python publish.py credits 1 6002
published order 6002 as message 11
$ python idempotent.py
order 6001: duplicate, skipped
order 6001: saved
order 6002: saved
rows in credits: 2Order 6001 went out as messages 9 and 10, and 6002 as message 11. The consumer prints order 6001: saved once and order 6001: duplicate, skipped once, then order 6002: saved. And rows in credits: 2.
Why not dedupe on the message ID? It catches redeliveries, but a publisher retry gets a new message ID. Only a business key, or an event ID the publisher sets once, catches both.
Keep the check and the side effect in one transaction. And when the side effect is outside your database, like a payment provider, pass the same key as its idempotency key.
Scenario 7: A payments consumer writes to a database and must never double-charge a customer. How do you design acks and writes?
Three layers:
- charges-eod - a pull subscription with exactly-once delivery
- One transaction - INSERT keyed by
payment_id - Ack after commit - checked with
ack_with_response()
Exactly-once delivery works on pull subscriptions only, within one region, by message ID: after a successful ack, that message isn't delivered again. The emulator accepts the setting and returns ack results, but it doesn't enforce the full guarantee.
Let's create the subscription with python create_eod.py, then publish four payment events with python publish_payments.py. One of them, pay-2, is sent twice, like a publisher retry.
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "charges-eod")
s = sub.create_subscription(request={
"name": path,
"topic": "projects/tb-shop-dev/topics/payments",
"enable_exactly_once_delivery": True})
print("charges-eod exactly-once:", s.enable_exactly_once_delivery)import json
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "payments")
payments = [("pay-1", 25.00), ("pay-2", 40.50),
("pay-2", 40.50), # a publisher retry
("pay-3", 34.50)]
for payment_id, amount in payments:
data = json.dumps({"payment_id": payment_id, "amount": amount})
message_id = pub.publish(topic, data.encode()).result()
print("published", payment_id, "as message", message_id)$ python create_eod.py
charges-eod exactly-once: True
$ python publish_payments.py
published pay-1 as message 12
published pay-2 as message 13
published pay-2 as message 14
published pay-3 as message 15We get charges-eod exactly-once: True. Then pay-2 is published as messages 13 and 14. Exactly-once can't help here: to Pub/Sub, those are two different messages.
Now the consumer. python charge.py inserts each charge with payment_id as the primary key, then acks with response. Then python total.py checks the table.
import json, sqlite3
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.subscriber import exceptions
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "charges-eod")
log = []
def callback(message):
pay = json.loads(message.data)
with sqlite3.connect("shop.db", timeout=10) as db: # one commit
cur = db.execute("INSERT INTO charges VALUES (?, ?)"
" ON CONFLICT (payment_id) DO NOTHING",
(pay["payment_id"], pay["amount"]))
done = "charged" if cur.rowcount else "duplicate, not charged"
try:
status = message.ack_with_response().result() # after commit
log.append(f"{pay['payment_id']}: {done}, ack {status.name}")
except exceptions.AcknowledgeError as e: # expect a redelivery
log.append(f"{pay['payment_id']}: ack {e.error_code.name}")
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=8)
except TimeoutError:
future.cancel()
future.result()
print("\n".join(sorted(log)))import sqlite3
with sqlite3.connect("shop.db") as db:
sql = "SELECT COUNT(*), SUM(amount) FROM charges"
n, total = db.execute(sql).fetchone()
print(f"charges in the database: {n}, total {total:.2f}")$ python charge.py
pay-1: charged, ack SUCCESS
pay-2: charged, ack SUCCESS
pay-2: duplicate, not charged, ack SUCCESS
pay-3: charged, ack SUCCESS
$ python total.py
charges in the database: 3, total 100.00pay-2: charged, ack SUCCESS once, and pay-2: duplicate, not charged, ack SUCCESS for the copy. Every ack returns SUCCESS. And charges in the database: 3, total 100.00.
If an ack fails, expect the message again, and the key makes the retry harmless. Pass the same payment ID to the payment provider too.
Exactly-once reduces duplicates. The key is what prevents a double charge.
Key points to remember
- Delivery is at least once: slow acks, lost acks and publisher retries
- Unacked messages come back after the ack deadline: nothing is lost in a crash
- Process, commit, then ack; never ack on receipt
- The client extends leases; nack only for transient problems
- Idempotent consumers keyed by a business key, not the message ID
The full script
This is a Python script that writes every file and runs every command above in order against the Pub/Sub emulator: run pip install google-cloud-pubsub first.
Show the full script that writes the files and runs every step
# ========================================================================
# Pub/Sub Interview Q&A #4 - Acknowledgements and redelivery
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Google Cloud Pub/Sub emulator (gcloud component, version 0.8.37)
# with google-cloud-pubsub 2.43.0; the outputs in the video and the blog post are the real outputs.
# To run it yourself (Linux, macOS or WSL):
# 1. gcloud components install pubsub-emulator then gcloud beta emulators pubsub start --project=tb-shop-dev
# 2. In a second terminal: export PUBSUB_EMULATOR_HOST=localhost:8085 PUBSUB_PROJECT_ID=tb-shop-dev
# 3. pip install google-cloud-pubsub then python PubSub_Interview_QA_04_Code.py in an empty folder.
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================
import os, subprocess, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: the starting files
write('schema.sql', """\
CREATE TABLE orders (order_id INTEGER PRIMARY KEY);
CREATE TABLE credits (order_id INTEGER PRIMARY KEY);
-- payment_id is the idempotency key
CREATE TABLE charges (payment_id TEXT PRIMARY KEY,
amount NUMERIC NOT NULL);
""")
write('setup.py', """\
import sqlite3
from google.cloud import pubsub_v1
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topics = {"orders": ["fulfil-pull"], "shipments": ["packing-pull"],
"ledger": ["ledger-pull"], "emails": ["email-pull"],
"reports": ["reports-stream", "reports-pull"],
"credits": ["credits-pull"], "payments": []}
for topic_id, subs in topics.items():
topic = pub.topic_path("tb-shop-dev", topic_id)
pub.create_topic(name=topic)
for sub_id in subs:
path = sub.subscription_path("tb-shop-dev", sub_id)
sub.create_subscription(name=path, topic=topic)
with sqlite3.connect("shop.db") as db:
db.executescript(open("schema.sql").read())
""")
write('publish.py', """\
import json, sys
from google.cloud import pubsub_v1
topic_id = sys.argv[1]
count, start = int(sys.argv[2]), int(sys.argv[3])
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", topic_id)
for order_id in range(start, start + count):
data = json.dumps({"order_id": order_id}).encode()
message_id = pub.publish(topic, data).result() # wait for each
print("published order", order_id, "as message", message_id)
""")
write('db.py', 'import os, sqlite3\n\n\ndef save_order(order_id):\n """Write one order. DB_DOWN=1 simulates a database outage."""\n if os.environ.get("DB_DOWN"):\n raise ConnectionError("database is down")\n with sqlite3.connect("shop.db") as db:\n db.execute("INSERT INTO orders VALUES (?)", (order_id,))\n\n\nif __name__ == "__main__":\n with sqlite3.connect("shop.db") as db:\n rows = db.execute("SELECT order_id FROM orders ORDER BY 1")\n print("orders in the database:", [r[0] for r in rows])\n')
run('python setup.py')
# Step 1: A slow worker: the ack deadline expires and a second worker gets the same order
write('slow_worker.py', """\
import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "fulfil-pull")
resp = sub.pull(subscription=path, max_messages=1) # worker A
a = resp.received_messages[0].message
print("worker A got message", a.message_id, a.data.decode())
time.sleep(13) # slow work: past the 10 s deadline
resp = sub.pull(subscription=path, max_messages=1) # worker B
rb = resp.received_messages[0]
b = rb.message
print("worker B got message", b.message_id, b.data.decode())
sub.acknowledge(subscription=path, ack_ids=[rb.ack_id])
print("same order processed twice:", a.data == b.data)
""")
run('python publish.py orders 1 1001')
run('python slow_worker.py')
# Step 2: A worker crashes mid-work; a new worker gets the unacked orders after the ack deadline
write('crash.py', """\
import os
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "packing-pull")
got = []
while len(got) < 3:
resp = sub.pull(subscription=path, max_messages=3)
got += resp.received_messages
for r in sorted(got, key=lambda r: r.message.data):
print("packing", r.message.data.decode())
if b"2002" in r.message.data:
print("crash! no ack for this order or the next")
os._exit(1) # the pod is killed
sub.acknowledge(subscription=path, ack_ids=[r.ack_id])
print(" acked")
""")
write('restart.py', """\
import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "packing-pull")
resp = sub.pull(subscription=path, max_messages=10, timeout=6)
n = len(resp.received_messages)
print("new worker, right away:", n, "messages")
got, start = [], time.time()
while len(got) < 2 and time.time() - start < 30:
resp = sub.pull(subscription=path, max_messages=10)
got += resp.received_messages
orders = sorted(r.message.data.decode() for r in got)
print("after the ack deadline:", orders)
sub.acknowledge(subscription=path, ack_ids=[r.ack_id for r in got])
""")
run('python publish.py shipments 3 2001', allow_error=True)
run('python crash.py', allow_error=True)
run('python restart.py', allow_error=True)
# Step 3: Ack first, then the database write fails: the order is lost
write('ack_first.py', """\
import json
from google.cloud import pubsub_v1
from db import save_order
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "ledger-pull")
resp = sub.pull(subscription=path, max_messages=10, timeout=10)
print("messages pulled:", len(resp.received_messages))
for r in resp.received_messages:
order_id = json.loads(r.message.data)["order_id"]
sub.acknowledge(subscription=path, ack_ids=[r.ack_id]) # too soon
print("acked order", order_id)
try:
save_order(order_id)
print("saved order", order_id)
except ConnectionError as e:
print("save failed:", e)
""")
run('python publish.py ledger 1 4001')
run('DB_DOWN=1 python ack_first.py')
run('python ack_first.py')
run('python db.py')
# Step 4: Write first, ack after: a failed write is nacked and retried
write('ack_after.py', """\
import json
from google.cloud import pubsub_v1
from db import save_order
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "ledger-pull")
resp = sub.pull(subscription=path, max_messages=10, timeout=10)
print("messages pulled:", len(resp.received_messages))
for r in resp.received_messages:
order_id = json.loads(r.message.data)["order_id"]
try:
save_order(order_id) # write first
sub.acknowledge(subscription=path, ack_ids=[r.ack_id])
print("saved, then acked order", order_id)
except ConnectionError as e:
sub.modify_ack_deadline(subscription=path, ack_ids=[r.ack_id],
ack_deadline_seconds=0) # a nack
print("save failed:", e, "- nacked order", order_id)
""")
run('python publish.py ledger 1 4002')
run('DB_DOWN=1 python ack_after.py')
run('python ack_after.py')
run('python db.py')
# Step 5: 25 seconds of work on a 10 second ack deadline: lease management extends it
write('long_job.py', """\
import time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import FlowControl
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "reports-stream")
deliveries = []
def callback(message):
deliveries.append(message.message_id)
time.sleep(25) # 25 s of work, 10 s ack deadline
message.ack()
flow = FlowControl(max_messages=1, # slow work: hold few
max_lease_duration=3600) # the default: 1 hour
future = sub.subscribe(path, callback=callback, flow_control=flow)
try:
future.result(timeout=30)
except TimeoutError:
future.cancel()
future.result()
print("ack deadline 10 s, work 25 s, deliveries:", len(deliveries))
""")
run('python publish.py reports 1 8001')
run('python long_job.py')
# Step 6: Unary pull: extend the deadline yourself with modify_ack_deadline
write('extend.py', """\
import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "reports-pull")
r = sub.pull(subscription=path, max_messages=1).received_messages[0]
sub.modify_ack_deadline(subscription=path, ack_ids=[r.ack_id],
ack_deadline_seconds=60) # 600 s at most
print("pulled message", r.message.message_id, "- deadline now 60 s")
time.sleep(15)
resp = sub.pull(subscription=path, max_messages=1, timeout=6)
print("pull 15 s later:", len(resp.received_messages), "messages")
sub.acknowledge(subscription=path, ack_ids=[r.ack_id])
print("acked when the work was done")
""")
run('python extend.py')
# Step 7: Nack twice, then ack
write('nack.py', """\
import collections
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "email-pull")
seen = collections.Counter()
def callback(message):
mid = message.message_id
seen[mid] += 1
if seen[mid] < 3:
print("delivery", seen[mid], "of message", mid, "- nack")
message.nack() # give it back now
else:
print("delivery", seen[mid], "of message", mid, "- ack")
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=5)
except TimeoutError:
future.cancel()
future.result()
""")
run('python publish.py emails 1 9001')
run('python nack.py')
# Step 8: A publisher retry and an idempotent consumer keyed by order_id
write('idempotent.py', 'import json, sqlite3\nfrom google.cloud import pubsub_v1\n\nsub = pubsub_v1.SubscriberClient()\npath = sub.subscription_path("tb-shop-dev", "credits-pull")\ndb, results = sqlite3.connect("shop.db"), []\nwhile True:\n resp = sub.pull(subscription=path, max_messages=10, timeout=10)\n if not resp.received_messages:\n break\n for r in resp.received_messages:\n order_id = json.loads(r.message.data)["order_id"]\n with db: # one transaction\n cur = db.execute("INSERT INTO credits VALUES (?)"\n " ON CONFLICT (order_id) DO NOTHING",\n (order_id,))\n done = "saved" if cur.rowcount else "duplicate, skipped"\n results.append(f"order {order_id}: {done}")\n sub.acknowledge(subscription=path, ack_ids=[r.ack_id])\nprint("\\n".join(sorted(results)))\nrows = db.execute("SELECT COUNT(*) FROM credits").fetchone()[0]\nprint("rows in credits:", rows)\n')
run('python publish.py credits 1 6001')
run('python publish.py credits 1 6001')
run('python publish.py credits 1 6002')
run('python idempotent.py')
# Step 9: An exactly-once pull subscription; four payment events, one of them a publisher retry
write('create_eod.py', """\
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "charges-eod")
s = sub.create_subscription(request={
"name": path,
"topic": "projects/tb-shop-dev/topics/payments",
"enable_exactly_once_delivery": True})
print("charges-eod exactly-once:", s.enable_exactly_once_delivery)
""")
write('publish_payments.py', """\
import json
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "payments")
payments = [("pay-1", 25.00), ("pay-2", 40.50),
("pay-2", 40.50), # a publisher retry
("pay-3", 34.50)]
for payment_id, amount in payments:
data = json.dumps({"payment_id": payment_id, "amount": amount})
message_id = pub.publish(topic, data.encode()).result()
print("published", payment_id, "as message", message_id)
""")
run('python create_eod.py')
run('python publish_payments.py')
# Step 10: Charge in one transaction keyed by payment_id, then ack_with_response
write('charge.py', 'import json, sqlite3\nfrom google.cloud import pubsub_v1\nfrom google.cloud.pubsub_v1.subscriber import exceptions\n\nsub = pubsub_v1.SubscriberClient()\npath = sub.subscription_path("tb-shop-dev", "charges-eod")\nlog = []\n\ndef callback(message):\n pay = json.loads(message.data)\n with sqlite3.connect("shop.db", timeout=10) as db: # one commit\n cur = db.execute("INSERT INTO charges VALUES (?, ?)"\n " ON CONFLICT (payment_id) DO NOTHING",\n (pay["payment_id"], pay["amount"]))\n done = "charged" if cur.rowcount else "duplicate, not charged"\n try:\n status = message.ack_with_response().result() # after commit\n log.append(f"{pay[\'payment_id\']}: {done}, ack {status.name}")\n except exceptions.AcknowledgeError as e: # expect a redelivery\n log.append(f"{pay[\'payment_id\']}: ack {e.error_code.name}")\n\nfuture = sub.subscribe(path, callback=callback)\ntry:\n future.result(timeout=8)\nexcept TimeoutError:\n future.cancel()\n future.result()\nprint("\\n".join(sorted(log)))\n')
write('total.py', """\
import sqlite3
with sqlite3.connect("shop.db") as db:
sql = "SELECT COUNT(*), SUM(amount) FROM charges"
n, total = db.execute(sql).fetchone()
print(f"charges in the database: {n}, total {total:.2f}")
""")
run('python charge.py')
run('python total.py')
Part 5: Ordering Keys and Exactly-Once
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 Google Cloud Pub/Sub emulator with google-cloud-pubsub 2.43.0. Features the emulator does not support are shown as code. Part of the Real Time Pub/Sub Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Pub/Sub documentation (cloud.google.com/pubsub/docs) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.