In this post we are going to cover ordering keys and exactly-once delivery 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: an inventory system needs updates per product in order and counted once, and you design it end to end.
By default, Pub/Sub gives no order and delivers at least once. Ordering keys add order per key, like per customer or per product. Exactly-once delivery stops redelivery after a successful ack. Interviewers ask about both because people often think they solve more than they do. If you can explain what each one really guarantees, what it costs, and where duplicates still come from, you show that you understand Pub/Sub at a senior level.
Part 5 of the Real Time Pub/Sub 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 ordering keys work?
- Question 2: A teammate turns on ordering and the publisher stops after one error. Why, and how do you recover?
- Question 3: Ordering slows everything down. What are the costs and limits of ordered delivery?
- Question 4: Your manager asks for exactly-once delivery. What does Pub/Sub exactly-once really guarantee?
- Question 5: With exactly-once on, the ack call returns an error. How do you handle ack results?
- Question 6: Someone says exactly-once in Pub/Sub means no duplicates anywhere. Why is that not true?
- Scenario 7: An inventory system needs updates per product in order and counted once. How do you design it end to end?
- The full script
The short answer
An ordering key gives in-order delivery per key, for messages published in the same region, on a subscription with message ordering on. A failed publish pauses the key until resume_publish. Exactly-once delivery stops redelivery after a successful ack, on pull subscriptions in one region, but publisher retries still create duplicates: keep consumers 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 the customer-updates topic in the project tb-shop-dev, and the profile-sync subscription with message ordering on.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "customer-updates")
pub.create_topic(name=topic)
sub.create_subscription(request={
"name": sub.subscription_path(project, "profile-sync"),
"topic": topic,
"enable_message_ordering": True, # set at creation
})from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import PublisherOptions
opts = PublisherOptions(enable_message_ordering=True)
pub = pubsub_v1.PublisherClient(publisher_options=opts)
topic = pub.topic_path("tb-shop-dev", "customer-updates")
futures = []
for v in range(1, 6):
for c in ["cust-A", "cust-B"]:
data = f"{c} v{v}".encode()
futures.append(pub.publish(topic, data, ordering_key=c))
for f in futures:
f.result(timeout=60)
print("published", len(futures), "updates with ordering keys")import threading
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "profile-sync")
seen, lock = {}, threading.Lock()
def callback(message):
with lock:
version = message.data.decode().split()[1]
seen.setdefault(message.ordering_key, []).append(version)
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=6)
except TimeoutError:
future.cancel()
future.result()
for key in sorted(seen):
print(key, "received", seen[key])from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient() # no ordering option
topic = pub.topic_path("tb-shop-dev", "customer-updates")
try:
pub.publish(topic, b"cust-A v6", ordering_key="cust-A")
except ValueError as e:
print("ValueError:", e)Question 1: Updates for one customer arrive out of order. How do ordering keys work?
An ordering key is a string we set on each message, like the customer ID. Messages with the same key, published in the same region, are delivered in the order Pub/Sub received them, to a subscription with message ordering on. Order is per key only. Messages without a key have no order.
Both sides opt in. The publisher sets enable_message_ordering in its publisher options. The subscription is created with enable_message_ordering true, which our setup did.
Let's run python publish.py. It publishes five versions each for cust-A and cust-B, interleaved. Then python consume.py reads them with streaming pull and records them by key.
$ python publish.py
published 10 updates with ordering keys
$ python consume.py
cust-A received ['v1', 'v2', 'v3', 'v4', 'v5']
cust-B received ['v1', 'v2', 'v3', 'v4', 'v5']published 10 updates with ordering keys. Then cust-A received ['v1', 'v2', 'v3', 'v4', 'v5'], and cust-B the same. Each customer gets v1 to v5, in order.
What if we forget the publisher option? Let's try with python plain.py.
$ python plain.py
ValueError: Cannot publish a message with an ordering key when message ordering is not enabled.The Python client refuses the key before anything is sent: ValueError: Cannot publish a message with an ordering key when message ordering is not enabled.
Key on the entity whose order matters: the customer, the order or the product. Not one key for everything, because that makes the topic a single lane. And publish each key through one region, with a locational or regional endpoint.
Question 2: A teammate turns on ordering and the publisher stops after one error. Why, and how do you recover?
That's by design. With ordering on, when a publish for a key fails with an error that can't be retried, the client pauses that key. It cancels the queued messages for it, and new publishes for it fail right away. Otherwise v2 could overtake v1.
Let's see it with python pause.py. The topic customer-updates-eu doesn't exist yet. We send v1 and v2 for cust-C. Then we fix the cause by creating the topic, call resume_publish with the topic and the key, and republish from the first failed message.
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import PublisherOptions
opts = PublisherOptions(enable_message_ordering=True)
pub = pubsub_v1.PublisherClient(publisher_options=opts)
topic = pub.topic_path("tb-shop-dev", "customer-updates-eu")
def send(version):
data = f"cust-C {version}".encode()
try:
future = pub.publish(topic, data, ordering_key="cust-C")
future.result(timeout=30)
print("cust-C", version, "published")
except Exception as e:
print("cust-C", version, "failed:", type(e).__name__, e)
send("v1") # the topic is not there yet
send("v2") # the key is paused now
pub.create_topic(name=topic) # fix the cause first
pub.resume_publish(topic, "cust-C")
send("v1") # republish from the first failure
send("v2")$ python pause.py
cust-C v1 failed: NotFound 404 Topic not found
cust-C v2 failed: PublishToPausedOrderingKeyException
cust-C v1 published
cust-C v2 publishedcust-C v1 failed: NotFound 404 Topic not found. And v2 isn't even sent: cust-C v2 failed: PublishToPausedOrderingKeyException. After the fix and the resume, cust-C v1 published and cust-C v2 published, in order.
The mistake I see: catching the error, logging it, and carrying on. The key stays paused, and every later update for that customer fails. Resume and republish, or stop and alert. Other keys keep publishing.
With ordering on, the Python client retries transient errors, like Unavailable, with no deadline. A pause comes from errors that won't fix themselves, like a missing topic or permission.
Question 3: Ordering slows everything down. What are the costs and limits of ordered delivery?
Ordering trades parallelism for order. Each key is a single lane.
| Limit | What it means |
|---|---|
| 1 MBps publish per ordering key | a hot key cannot go faster |
| Pull: one outstanding batch per key | the next batch waits for the acks |
| Push: one outstanding message per key | one message in flight per key |
So one slow message holds back its own key. Let's publish the ten updates again and run python slow_key.py, where cust-A v3 takes three seconds.
import threading, time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "profile-sync")
done, lock = [], threading.Lock()
def callback(message):
if message.data == b"cust-A v3":
time.sleep(3) # one slow update
with lock:
done.append(message.data.decode())
message.ack()
future = sub.subscribe(path, callback=callback)
time.sleep(8)
future.cancel()
future.result()
a3 = done.index("cust-A v3")
for key in ["cust-A", "cust-B"]:
early = [d.split()[1] for d in done[:a3] if d.startswith(key)]
print(key, "done before cust-A v3:", early)
print("done after cust-A v3:", done[a3 + 1:])$ python publish.py
published 10 updates with ordering keys
$ python slow_key.py
cust-A done before cust-A v3: ['v1', 'v2']
cust-B done before cust-A v3: ['v1', 'v2', 'v3', 'v4', 'v5']
done after cust-A v3: ['cust-A v4', 'cust-A v5']Before cust-A v3 finished, cust-A had done only ['v1', 'v2'], but cust-B finished ['v1', 'v2', 'v3', 'v4', 'v5']. And done after cust-A v3: ['cust-A v4', 'cust-A v5']. Customer A's v4 and v5 waited for v3. Customer B didn't.
Redelivery is the other cost. On the real service, if one message for a key is redelivered, every later message for that key is redelivered too, even ones you already acked. So consumers must handle repeats.
And it's fixed at creation. Let's try to turn ordering off on profile-sync with python immutable.py.
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "profile-sync")
change = {"name": path, "enable_message_ordering": False}
try:
sub.update_subscription(request={
"subscription": change,
"update_mask": {"paths": ["enable_message_ordering"]},
}, timeout=30)
except Exception as e:
print(type(e).__name__, e)$ python immutable.py
InvalidArgument 400 Invalid update_mask provided in the UpdateSubscriptionRequest: the
enable_message_ordering field in the Subscription is not mutable.We get InvalidArgument 400: the enable_message_ordering field in the Subscription is not mutable. To change it, create a new subscription.
So keep these in mind:
- Hot keys - monitor oldest_unacked_message_age_by_region
- Dataflow - don't enable message ordering
- Dead-letter topic - the order can break when a message moves out
- Only where the business needs order
Question 4: Your manager asks for exactly-once delivery. What does Pub/Sub exactly-once really guarantee?
Exactly-once delivery is a subscription setting. When it's on:
- No redelivery after a successful ack - and you learn if the ack succeeded
- No redelivery while outstanding - inside the ack deadline
- Only the latest ack ID is valid
The scope is precise. Pull and streaming pull subscriptions only, not push or export. It's guaranteed within one cloud region, based on the Pub/Sub message ID. Unlike ordering, you can turn it on for an existing subscription.
On Google Cloud, it's one flag on gcloud.
gcloud pubsub subscriptions create inventory-eod \
--topic=inventory \
--enable-exactly-once-deliveryLet's do the same with python eod_setup.py, which creates inventory-eod and reads it back.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "inventory")
path = sub.subscription_path(project, "inventory-eod")
pub.create_topic(name=topic)
sub.create_subscription(request={
"name": path, "topic": topic,
"enable_exactly_once_delivery": True,
})
s = sub.get_subscription(subscription=path)
print(s.name)
print("enable_exactly_once_delivery:", s.enable_exactly_once_delivery)$ python eod_setup.py
projects/tb-shop-dev/subscriptions/inventory-eod
enable_exactly_once_delivery: TrueIt prints projects/tb-shop-dev/subscriptions/inventory-eod and enable_exactly_once_delivery: True.
One honest note: the emulator accepts the setting and runs the ack calls, but it doesn't enforce the full semantics. So the demo shows the API, and the guarantee comes from the real service.
Higher publish to subscribe latency, and separate exactly-once quotas per region. Use streaming pull for throughput, and turn it on only where a redelivery is expensive.
Question 5: With exactly-once on, the ack call returns an error. How do you handle ack results?
Use ack_with_response(). With exactly-once on, an ack is a confirmed operation, and the future tells you how it went. SUCCESS means the message won't be redelivered. A failure means: expect it again.
Let's publish one stock event and run python ack.py. The consumer acks the same message twice.
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.subscriber import exceptions
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "inventory")
path = sub.subscription_path("tb-shop-dev", "inventory-eod")
pub.publish(topic, b"sku-42 -1", event_id="e-501").result()
def callback(message):
for attempt in ["first ack", "second ack"]:
try:
status = message.ack_with_response().result(timeout=30)
print(attempt, "->", status.name)
except exceptions.AcknowledgeError as e:
print(attempt, "->", type(e).__name__, e.error_code.name)
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=5)
except TimeoutError:
future.cancel()
future.result()$ python ack.py
first ack -> SUCCESS
second ack -> AcknowledgeError INVALID_ACK_IDfirst ack -> SUCCESS, and second ack -> AcknowledgeError INVALID_ACK_ID.
On the real service, you see that code in a different situation.
| INVALID_ACK_ID | When |
|---|---|
| Real service | ack ID expired, or replaced by a newer delivery |
| Emulator | a double ack; an expired ack ID is not rejected |
The emulator doesn't reject an expired ack ID, so we show the code with a double ack.
The Python client already retries transient ack failures. So when the future raises, don't retry the ack yourself. Treat the message as not acked, and make the redelivery harmless: your write already happened.
Nack and modify ack deadline have with-response versions too. With exactly-once, the Python client extends leases by at least 60 seconds. And I alert on the exactly_once_warning_count metric.
Question 6: Someone says exactly-once in Pub/Sub means no duplicates anywhere. Why is that not true?
Because exactly-once covers delivery from one subscription, not your whole pipeline. It works on the Pub/Sub message ID, inside one region. Anything that creates a new message ID is a new message.
The classic case is a publisher retry: the publish worked, but the response was lost, so the publisher sends again. Let's simulate that with python dedupe.py. It publishes the same event, e-701, twice, and reads the exactly-once subscription.
import threading
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "inventory")
for _ in range(2): # a retry after a lost publish response
pub.publish(topic, b"sku-42 -1", event_id="e-701").result()
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "inventory-eod")
ids, done, lock = [], set(), threading.Lock()
def callback(message):
with lock:
ids.append(message.message_id)
done.add(message.attributes["event_id"]) # business key
message.ack_with_response().result(timeout=30)
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=5)
except TimeoutError:
future.cancel()
future.result()
print("deliveries:", len(ids), "message_ids:", sorted(ids))
print("events processed:", sorted(done))$ python dedupe.py
deliveries: 2 message_ids: ['24', '25']
events processed: ['e-701']The subscription delivers both: deliveries: 2 message_ids: ['24', '25']. Two message IDs, one event ID. Because the consumer keys on the event ID, we get events processed: ['e-701']. Two deliveries, one event processed. In production, that's a unique key or a merge in the database.
There are more gaps:
- Publisher retries - new message ID
- Other regions - subscribers there can get duplicates
- Your side effects - DB writes, emails
- BigQuery, Cloud Storage subs - no exactly-once
Exactly-once reduces redeliveries, and idempotency removes duplicates. You need the second, even when you have the first.
Scenario 7: An inventory system needs updates per product in order and counted once. How do you design it end to end?
Here is the design:
- Ordering key = SKU - one region, locational endpoint
- Pull subscription - ordering + exactly-once, same region, acks in order
- One transaction - dedupe row + stock change, ack after the commit
- Dead-letter topic - can break the order for that key
- Alert - oldest_unacked_message_age_by_region
Order comes from the key: the SKU. Updates for one product stay in sequence, and different products run in parallel. Publishers use a locational endpoint, like us-east1, so each key goes through one region.
# all publishes for a key through one region
pub = pubsub_v1.PublisherClient(
publisher_options=opts,
client_options={
"api_endpoint": "us-east1-pubsub.googleapis.com:443"},
)Consumers use the same locational endpoint, because exactly-once holds per region. Let's create the subscription with both settings on with python stock_setup.py.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "stock-events")
pub.create_topic(name=topic)
s = sub.create_subscription(request={
"name": sub.subscription_path(project, "stock-apply"),
"topic": topic,
"enable_message_ordering": True,
"enable_exactly_once_delivery": True,
})
print(s.name.split("/")[-1], "ordering:", s.enable_message_ordering,
"exactly-once:", s.enable_exactly_once_delivery)$ python stock_setup.py
stock-apply ordering: True exactly-once: Truestock-apply ordering: True exactly-once: True. With both on, acks go in order, so a client tops out around thousands of messages a second.
Counted once comes from the consumer. In one database transaction, we insert the event ID into an applied table, and change the stock only if it's new. Only after the commit do we ack, with ack_with_response(). We publish six events with python stock_publish.py, including e2 twice as a publisher retry, and then run python stock_apply.py.
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import PublisherOptions
opts = PublisherOptions(enable_message_ordering=True)
pub = pubsub_v1.PublisherClient(publisher_options=opts)
topic = pub.topic_path("tb-shop-dev", "stock-events")
events = [("sku-1", "e1", 10), ("sku-2", "e4", 4),
("sku-1", "e2", -3),
("sku-1", "e2", -3), # the publisher retried: a duplicate
("sku-2", "e5", -4), ("sku-1", "e3", -2)]
for sku, event_id, delta in events:
pub.publish(topic, str(delta).encode(), ordering_key=sku,
event_id=event_id).result(timeout=60)
print("published", len(events), "stock events")import sqlite3, threading, time
from google.cloud import pubsub_v1
db = sqlite3.connect(":memory:", check_same_thread=False)
db.execute("create table stock (sku text primary key, qty int)")
db.execute("create table applied (event_id text primary key)")
UPSERT = ("insert into stock values (?, ?) on conflict (sku)"
" do update set qty = qty + excluded.qty returning qty")
lock, levels, skipped = threading.Lock(), {}, []
def apply(message):
sku, event = message.ordering_key, message.attributes["event_id"]
with lock, db: # one transaction per event
if db.execute("insert or ignore into applied values (?)",
(event,)).rowcount:
cur = db.execute(UPSERT, (sku, int(message.data)))
levels.setdefault(sku, []).append(cur.fetchone()[0])
else:
skipped.append(event) # seen before: skip it
message.ack_with_response().result(timeout=30) # after the commit
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "stock-apply")
future = sub.subscribe(path, callback=apply)
time.sleep(6)
future.cancel()
future.result()
for sku in sorted(levels):
print(sku, "stock levels", levels[sku])
print("duplicates skipped:", skipped)$ python stock_publish.py
published 6 stock events
$ python stock_apply.py
sku-1 stock levels [10, 7, 5]
sku-2 stock levels [4, 0]
duplicates skipped: ['e2']sku-1 stock levels [10, 7, 5], in order. sku-2 stock levels [4, 0]. And duplicates skipped: ['e2'].
Ordering gives the sequence, exactly-once cuts redeliveries, and the idempotent write counts each event once.
Key points to remember
- Ordering key + same region + ordered subscription = order per key
- A failed publish pauses the key: fix, resume_publish, republish
- Ordering costs: 1 MBps per key, head-of-line waits, fixed at creation
- Exactly-once: no redelivery after a successful ack, pull only, one region
- Publisher retries still duplicate: dedupe by a business key
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 #5 - Ordering keys and exactly-once delivery
# 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_05_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('setup.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "customer-updates")
pub.create_topic(name=topic)
sub.create_subscription(request={
"name": sub.subscription_path(project, "profile-sync"),
"topic": topic,
"enable_message_ordering": True, # set at creation
})
""")
write('publish.py', """\
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import PublisherOptions
opts = PublisherOptions(enable_message_ordering=True)
pub = pubsub_v1.PublisherClient(publisher_options=opts)
topic = pub.topic_path("tb-shop-dev", "customer-updates")
futures = []
for v in range(1, 6):
for c in ["cust-A", "cust-B"]:
data = f"{c} v{v}".encode()
futures.append(pub.publish(topic, data, ordering_key=c))
for f in futures:
f.result(timeout=60)
print("published", len(futures), "updates with ordering keys")
""")
write('consume.py', """\
import threading
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "profile-sync")
seen, lock = {}, threading.Lock()
def callback(message):
with lock:
version = message.data.decode().split()[1]
seen.setdefault(message.ordering_key, []).append(version)
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=6)
except TimeoutError:
future.cancel()
future.result()
for key in sorted(seen):
print(key, "received", seen[key])
""")
write('plain.py', """\
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient() # no ordering option
topic = pub.topic_path("tb-shop-dev", "customer-updates")
try:
pub.publish(topic, b"cust-A v6", ordering_key="cust-A")
except ValueError as e:
print("ValueError:", e)
""")
run('python setup.py')
# Step 1: Publish five updates per customer with ordering keys, then read them from the ordered subscription
run('python publish.py')
run('python consume.py')
# Step 2: An ordering key on a publisher without message ordering
run('python plain.py')
# Step 3: A failed publish pauses the ordering key; fix the cause, resume_publish, republish
write('pause.py', """\
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import PublisherOptions
opts = PublisherOptions(enable_message_ordering=True)
pub = pubsub_v1.PublisherClient(publisher_options=opts)
topic = pub.topic_path("tb-shop-dev", "customer-updates-eu")
def send(version):
data = f"cust-C {version}".encode()
try:
future = pub.publish(topic, data, ordering_key="cust-C")
future.result(timeout=30)
print("cust-C", version, "published")
except Exception as e:
print("cust-C", version, "failed:", type(e).__name__, e)
send("v1") # the topic is not there yet
send("v2") # the key is paused now
pub.create_topic(name=topic) # fix the cause first
pub.resume_publish(topic, "cust-C")
send("v1") # republish from the first failure
send("v2")
""")
run('python pause.py')
# Step 4: One slow message holds back its own key, not the others
write('slow_key.py', """\
import threading, time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "profile-sync")
done, lock = [], threading.Lock()
def callback(message):
if message.data == b"cust-A v3":
time.sleep(3) # one slow update
with lock:
done.append(message.data.decode())
message.ack()
future = sub.subscribe(path, callback=callback)
time.sleep(8)
future.cancel()
future.result()
a3 = done.index("cust-A v3")
for key in ["cust-A", "cust-B"]:
early = [d.split()[1] for d in done[:a3] if d.startswith(key)]
print(key, "done before cust-A v3:", early)
print("done after cust-A v3:", done[a3 + 1:])
""")
run('python publish.py')
run('python slow_key.py')
# Step 5: Message ordering cannot be changed on an existing subscription
write('immutable.py', """\
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "profile-sync")
change = {"name": path, "enable_message_ordering": False}
try:
sub.update_subscription(request={
"subscription": change,
"update_mask": {"paths": ["enable_message_ordering"]},
}, timeout=30)
except Exception as e:
print(type(e).__name__, e)
""")
run('python immutable.py')
# Step 6: A pull subscription with exactly-once delivery
write('eod_setup.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "inventory")
path = sub.subscription_path(project, "inventory-eod")
pub.create_topic(name=topic)
sub.create_subscription(request={
"name": path, "topic": topic,
"enable_exactly_once_delivery": True,
})
s = sub.get_subscription(subscription=path)
print(s.name)
print("enable_exactly_once_delivery:", s.enable_exactly_once_delivery)
""")
run('python eod_setup.py')
# Step 7: ack_with_response: the first ack succeeds, a second ack of the same message fails
write('ack.py', """\
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.subscriber import exceptions
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "inventory")
path = sub.subscription_path("tb-shop-dev", "inventory-eod")
pub.publish(topic, b"sku-42 -1", event_id="e-501").result()
def callback(message):
for attempt in ["first ack", "second ack"]:
try:
status = message.ack_with_response().result(timeout=30)
print(attempt, "->", status.name)
except exceptions.AcknowledgeError as e:
print(attempt, "->", type(e).__name__, e.error_code.name)
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=5)
except TimeoutError:
future.cancel()
future.result()
""")
run('python ack.py')
# Step 8: A publisher retry creates a second message: exactly-once delivers both
write('dedupe.py', """\
import threading
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "inventory")
for _ in range(2): # a retry after a lost publish response
pub.publish(topic, b"sku-42 -1", event_id="e-701").result()
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "inventory-eod")
ids, done, lock = [], set(), threading.Lock()
def callback(message):
with lock:
ids.append(message.message_id)
done.add(message.attributes["event_id"]) # business key
message.ack_with_response().result(timeout=30)
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=5)
except TimeoutError:
future.cancel()
future.result()
print("deliveries:", len(ids), "message_ids:", sorted(ids))
print("events processed:", sorted(done))
""")
run('python dedupe.py')
# Step 9: Scenario: an ordered, exactly-once pull subscription for stock events
write('stock_setup.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "stock-events")
pub.create_topic(name=topic)
s = sub.create_subscription(request={
"name": sub.subscription_path(project, "stock-apply"),
"topic": topic,
"enable_message_ordering": True,
"enable_exactly_once_delivery": True,
})
print(s.name.split("/")[-1], "ordering:", s.enable_message_ordering,
"exactly-once:", s.enable_exactly_once_delivery)
""")
run('python stock_setup.py')
# Step 10: Scenario: publish keyed stock events, apply each once in a database transaction, ack after the commit
write('stock_publish.py', """\
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import PublisherOptions
opts = PublisherOptions(enable_message_ordering=True)
pub = pubsub_v1.PublisherClient(publisher_options=opts)
topic = pub.topic_path("tb-shop-dev", "stock-events")
events = [("sku-1", "e1", 10), ("sku-2", "e4", 4),
("sku-1", "e2", -3),
("sku-1", "e2", -3), # the publisher retried: a duplicate
("sku-2", "e5", -4), ("sku-1", "e3", -2)]
for sku, event_id, delta in events:
pub.publish(topic, str(delta).encode(), ordering_key=sku,
event_id=event_id).result(timeout=60)
print("published", len(events), "stock events")
""")
write('stock_apply.py', """\
import sqlite3, threading, time
from google.cloud import pubsub_v1
db = sqlite3.connect(":memory:", check_same_thread=False)
db.execute("create table stock (sku text primary key, qty int)")
db.execute("create table applied (event_id text primary key)")
UPSERT = ("insert into stock values (?, ?) on conflict (sku)"
" do update set qty = qty + excluded.qty returning qty")
lock, levels, skipped = threading.Lock(), {}, []
def apply(message):
sku, event = message.ordering_key, message.attributes["event_id"]
with lock, db: # one transaction per event
if db.execute("insert or ignore into applied values (?)",
(event,)).rowcount:
cur = db.execute(UPSERT, (sku, int(message.data)))
levels.setdefault(sku, []).append(cur.fetchone()[0])
else:
skipped.append(event) # seen before: skip it
message.ack_with_response().result(timeout=30) # after the commit
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "stock-apply")
future = sub.subscribe(path, callback=apply)
time.sleep(6)
future.cancel()
future.result()
for sku in sorted(levels):
print(sku, "stock levels", levels[sku])
print("duplicates skipped:", skipped)
""")
run('python stock_publish.py')
run('python stock_apply.py')
Part 6: Dead-Letter Topics and Retries
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.