In this post we are going to cover real-world manager scenarios for 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: in the final round, the hiring manager asks us to design an event platform on Google Cloud for an online retailer.
This is the last part of the series, and it pulls everything together. A manager doesn't ask what a subscription is. They ask why the backlog grows every evening, how we recover from a bad release, who can publish to a topic, and why the bill jumped. Interviewers ask these to see if we can run Pub/Sub in production, not just describe it.
Part 10 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 subscription backlog grows every evening at peak. How do you investigate and fix it?
- Question 2: A consumer bug acked a day of messages without processing them. How do you recover?
- Question 3: Leadership asks how to secure topics so only the right services publish and read. What do you set up?
- Question 4: The team needs events in two regions for disaster recovery. How do you design it?
- Question 5: A partner outside the company must send events to your topic. How do you do it safely?
- Question 6: Pub/Sub costs jumped this month. Where do you look first?
- Scenario 7: In the final round, the hiring manager asks you to design an event platform on Google Cloud for an online retailer. What is your design?
- The full script
The short answer
Fix a growing backlog by reading the by-region backlog metrics and scaling subscribers; recover from bad acks with seek, which needs retained acked messages or topic retention. Secure with least-privilege IAM per topic and subscription, plan two-region failover in the clients, and give partners their own topic with a dead-letter topic.
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. What the emulator doesn't support, like IAM and metrics, is shown as code and tables. We run python setup.py once before the first step. It creates the orders topic and the pull subscription orders-fulfilment, and backlog.txt starts at 0.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path(project, "orders")
publisher.create_topic(name=topic)
subscriber.create_subscription(
name=subscriber.subscription_path(project, "orders-fulfilment"),
topic=topic)0Question 1: The subscription backlog grows every evening at peak. How do you investigate and fix it?
Start with the metrics, the by-region versions.
| Metric (Cloud Monitoring, subscription/...) | Tells you |
|---|---|
| num_unacked_messages_by_region | the backlog |
| oldest_unacked_message_age_by_region | how late you are |
| expired_ack_deadlines_count | workers hold more than they finish |
Unacked messages is the backlog, and oldest unacked message age is how late we are. Expired ack deadlines means workers hold more than they can finish, so messages are redelivered and work is done twice.
The emulator has no metrics, so evening.py counts published minus acked. At peak, the website publishes 30 orders a tick, and one worker handles 10. We run python evening.py 30 1 5.
import sys
from google.cloud import pubsub_v1
incoming, workers, ticks = (int(a) for a in sys.argv[1:])
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path("tb-shop-dev", "orders")
sub = subscriber.subscription_path("tb-shop-dev", "orders-fulfilment")
backlog = int(open("backlog.txt").read()) # published minus acked
for tick in range(1, ticks + 1):
for i in range(incoming): # the website publishes
publisher.publish(topic, b'{"order": %d}' % i).result()
acked = 0
for w in range(workers): # 10 per worker
resp = subscriber.pull(subscription=sub, max_messages=10,
timeout=10)
ack_ids = [m.ack_id for m in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
acked += len(ack_ids)
backlog += incoming - acked
print(f"tick {tick}: in {incoming}, acked {acked},",
f"backlog {backlog}")
open("backlog.txt", "w").write(str(backlog))$ python evening.py 30 1 5
tick 1: in 30, acked 10, backlog 20
tick 2: in 30, acked 10, backlog 40
tick 3: in 30, acked 10, backlog 60
tick 4: in 30, acked 10, backlog 80
tick 5: in 30, acked 10, backlog 100The backlog grows by 20 every tick: backlog 20, 40, 60, 80, and backlog 100 after tick 5.
The fix here is to scale out: six workers, simulated as six pulls of 10, share the same subscription. We run python evening.py 10 6 2 and python check.py.
from google.api_core.exceptions import DeadlineExceeded
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "orders-fulfilment")
try:
resp = subscriber.pull(subscription=sub, max_messages=10,
timeout=5)
left = len(resp.received_messages)
except DeadlineExceeded: # nothing to deliver
left = 0
print("messages still waiting in orders-fulfilment:", left)$ python evening.py 10 6 2
tick 1: in 10, acked 60, backlog 50
tick 2: in 10, acked 60, backlog 0
$ python check.py
messages still waiting in orders-fulfilment: 0Each tick the six workers ack 60, so the backlog goes to 50, then 0. A final pull finds messages still waiting in orders-fulfilment: 0.
Scaling out is the first fix, but not the only cause to check:
- Scale out - workers share one subscription
- Flow control and threads - Python client: 1,000 messages, 10 threads
- Slow sink - the database behind the consumer
- Hot ordering key - or a quota
Alert on oldest unacked message age, and autoscale workers on the backlog before the evening starts.
Question 2: A consumer bug acked a day of messages without processing them. How do you recover?
Seek the subscription back to a time before the bug. Messages published before that time count as acked, everything after is delivered again. But only if Pub/Sub still has them: the subscription must retain acked messages, or the topic must have message retention, covering that day.
Let's create the payments-ledger subscription with retained acked messages, publish three payments, and process them with consumer v1. We run python create_ledger.py, python pay.py 1 3 and python consumer.py v1.
from google.cloud import pubsub_v1
from google.protobuf import duration_pb2
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
sub = subscriber.subscription_path("tb-shop-dev", "payments-ledger")
publisher.create_topic(name=topic)
s = subscriber.create_subscription(request={
"name": sub, "topic": topic,
"retain_acked_messages": True, # keep acked messages
"message_retention_duration": duration_pb2.Duration(
seconds=7 * 24 * 3600)}) # for 7 days
print("retain_acked_messages:", s.retain_acked_messages,
"| retention:", s.message_retention_duration)import sys
from google.cloud import pubsub_v1
first, last = int(sys.argv[1]), int(sys.argv[2])
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
names = [f"pay-{i}" for i in range(first, last + 1)]
for name in names:
publisher.publish(topic, name.encode()).result()
print("published", ", ".join(names))import sys
from google.cloud import pubsub_v1
version = sys.argv[1]
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "payments-ledger")
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
processed = []
for m in resp.received_messages:
if version != "v2": # v2 bug: acks, never processes
processed.append(m.message.data.decode())
ack_ids = [m.ack_id for m in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
print(version, "acked", len(ack_ids),
"| processed", sorted(processed))$ python create_ledger.py
retain_acked_messages: True | retention: 7 days, 0:00:00
$ python pay.py 1 3
published pay-1, pay-2, pay-3
$ python consumer.py v1
v1 acked 3 | processed ['pay-1', 'pay-2', 'pay-3']It prints retain_acked_messages: True | retention: 7 days, 0:00:00, and v1 processes pay-1 to pay-3. Now version 2 is released. We run python release.py, python pay.py 4 8 and python consumer.py v2.
import time
from datetime import datetime, timezone
time.sleep(1.5) # margin before the release
with open("release.txt", "w") as f:
f.write(datetime.now(timezone.utc).isoformat())
time.sleep(1.5)
print("consumer v2 released, release time saved")$ python release.py
consumer v2 released, release time saved
$ python pay.py 4 8
published pay-4, pay-5, pay-6, pay-7, pay-8
$ python consumer.py v2
v2 acked 5 | processed []v2 acked 5 | processed []. It acked five payments and processed none of them. The fix: seek to the release time with python seek.py, and deploy version 3 with python consumer.py v3.
from datetime import datetime
from google.cloud import pubsub_v1
from google.protobuf import timestamp_pb2
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "payments-ledger")
release = datetime.fromisoformat(open("release.txt").read())
ts = timestamp_pb2.Timestamp()
ts.FromDatetime(release)
subscriber.seek(request={"subscription": sub, "time": ts})
print("payments-ledger rewound to the v2 release time")$ python seek.py
payments-ledger rewound to the v2 release time
$ python consumer.py v3
v3 acked 5 | processed ['pay-4', 'pay-5', 'pay-6', 'pay-7', 'pay-8']Payments pay-4 to pay-8 are delivered again and processed. Payments 1 to 3 are not, because they came before the release.
On the real service, seek is eventually consistent, so give it a minute. And publish times are server times, so seek a little earlier and keep consumers idempotent.
Take a snapshot before every consumer deploy. Then a bad release is one seek to that snapshot.
Question 3: Leadership asks how to secure topics so only the right services publish and read. What do you set up?
Least privilege, on each topic and subscription, not on the project. Every service gets its own service account.
| Who | Role | Granted on |
|---|---|---|
| checkout service | roles/pubsub.publisher | the orders topic |
| billing service | roles/pubsub.subscriber | the orders-billing subscription |
| on-call engineers | roles/pubsub.viewer | the project |
| platform pipeline (Terraform) | roles/pubsub.admin | the project |
The checkout service gets Pub/Sub Publisher on the orders topic only. The billing service gets Pub/Sub Subscriber on its own subscription only. On-call engineers get Viewer, and only the platform pipeline that runs Terraform gets Admin, because it also manages these bindings. The Subscriber role can't create subscriptions, which is what we want: apps consume, the platform creates.
It's two gcloud commands, shown as code only: the emulator has no IAM at all.
# code only: the emulator has no IAM
CHECKOUT=serviceAccount:checkout@tb-shop-prod.iam.gserviceaccount.com
BILLING=serviceAccount:billing@tb-shop-prod.iam.gserviceaccount.com
gcloud pubsub topics add-iam-policy-binding orders \
--member=$CHECKOUT --role=roles/pubsub.publisher
gcloud pubsub subscriptions add-iam-policy-binding orders-billing \
--member=$BILLING --role=roles/pubsub.subscriberIf policy requires customer-managed keys, the Pub/Sub service agent needs roles/cloudkms.cryptoKeyEncrypterDecrypter. If the key becomes unavailable, publishing and delivery fail with FAILED_PRECONDITION. Push subscriptions authenticate with OIDC tokens.
One shared service account with Editor on the whole project. Nobody can tell who published what, and any service can delete a subscription.
Question 4: The team needs events in two regions for disaster recovery. How do you design it?
Pub/Sub is a global service: topics and subscriptions aren't tied to a region. But each message is stored in the region it was published in, and Pub/Sub never fails over by itself. So we design the clients for two regions.
First, the endpoints. Google's reliability guide says avoid the global endpoint here. Inside Google Cloud, it sends requests to the caller's own region, and doesn't re-route them when Pub/Sub there is down. So we pin clients with locational or regional endpoints.
| Endpoint | Example | Routing |
|---|---|---|
| Global (avoid for DR) | pubsub.googleapis.com | in Google Cloud: caller's region, not re-routed |
| Locational | us-central1-pubsub.googleapis.com | almost always the named region |
| Regional | pubsub.us-central1.rep.googleapis.com | never re-routed; needs enforce in transit |
Then pick a pattern.
| Pattern | Publishers | If a region fails | Cost |
|---|---|---|---|
| Primary + fallback | publish once, switch on errors | its stored messages wait | no extra |
| Active-active | publish to both regions | the other region delivers | 2x throughput, duplicates |
Primary and fallback: publish once to the primary, switch publishers to the fallback region on errors, and keep subscribers connected to both. Cheap, but messages already stored in the failed region wait until it recovers. Active-active: publish every message through both regions. Nothing waits for a region, but throughput cost doubles and every message arrives twice, so consumers must be idempotent.
A message storage policy, if we set one, must allow both regions. Here is the config, code only.
# code only: allow both regions, one client per region
gcloud pubsub topics create orders \
--message-storage-policy-allowed-regions=us-east1,us-central1
east = pubsub_v1.PublisherClient(client_options={
"api_endpoint": "us-east1-pubsub.googleapis.com:443"})
central = pubsub_v1.PublisherClient(client_options={
"api_endpoint": "us-central1-pubsub.googleapis.com:443"})Run a game day that fails publishers over to the second region, and track the backlog per region with the by-region metrics.
Question 5: A partner outside the company must send events to your topic. How do you do it safely?
Give the partner its own ingress topic, never our internal topics, and grant their service account Pub/Sub Publisher on that one topic. If their data already sits in Kinesis, Event Hubs, Amazon MSK, Confluent Cloud or Cloud Storage, an import topic can pull it in instead.
Then plan for bad events. A schema on the topic rejects malformed data. Without one, as here, the safety net is a dead-letter topic after five attempts, with its own subscription, or dead letters are lost. We run python create_partner.py.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path(project, "partner-orders")
dlq = publisher.topic_path(project, "partner-orders-dlq")
for t in (topic, dlq):
publisher.create_topic(name=t)
sub_path = subscriber.subscription_path
subscriber.create_subscription( # so dead letters are kept
name=sub_path(project, "partner-orders-dlq-sub"), topic=dlq)
s = subscriber.create_subscription(request={
"name": sub_path(project, "partner-orders-sub"),
"topic": topic,
"dead_letter_policy": {"dead_letter_topic": dlq,
"max_delivery_attempts": 5}})
p = s.dead_letter_policy
print("dead letter topic:", p.dead_letter_topic)
print("max delivery attempts:", p.max_delivery_attempts)$ python create_partner.py
dead letter topic: projects/tb-shop-dev/topics/partner-orders-dlq
max delivery attempts: 5The subscription has dead letter topic: projects/tb-shop-dev/topics/partner-orders-dlq and max delivery attempts: 5. Now the partner sends three orders, and P-2 is broken JSON. We run python partner_send.py and python partner_consumer.py.
from google.cloud import pubsub_v1
orders = ['{"order": "P-1", "amount": 40}',
'{"order": "P-2", "amount": ', # broken JSON
'{"order": "P-3", "amount": 15}']
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "partner-orders")
for o in orders:
publisher.publish(topic, o.encode(), partner="acme").result()
print("acme sent", len(orders), "orders")import json
from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = "projects/tb-shop-dev/subscriptions/partner-orders-sub"
done, attempts = [], []
def callback(m):
try:
done.append(json.loads(m.data)["order"])
m.ack()
except ValueError: # malformed: give it back
attempts.append(m.delivery_attempt)
m.nack()
future = subscriber.subscribe(sub, callback=callback)
try:
future.result(timeout=8)
except TimeoutError:
future.cancel()
future.result()
print("processed", sorted(done))
print("broken JSON nacked, delivery attempts", sorted(attempts))from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = "projects/tb-shop-dev/subscriptions/partner-orders-dlq-sub"
resp = subscriber.pull(subscription=sub, max_messages=10, timeout=10)
for r in resp.received_messages:
a = r.message.attributes
print("dead letter:", r.message.data.decode())
print(" partner:", a["partner"], "| delivery count:",
a["CloudPubSubDeadLetterSourceDeliveryCount"])
print(" source:", a["CloudPubSubDeadLetterSourceSubscription"])
ack_ids = [r.ack_id for r in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)$ python partner_send.py
acme sent 3 orders
$ python partner_consumer.py
processed ['P-1', 'P-3']
broken JSON nacked, delivery attempts [1, 2, 3, 4, 5]The consumer processes ['P-1', 'P-3'], and nacks P-2 on delivery attempts [1, 2, 3, 4, 5]. Then it leaves the subscription. Let's read the dead-letter subscription with python read_dlq.py.
$ python read_dlq.py
dead letter: {"order": "P-2", "amount":
partner: acme | delivery count: 5
source: partner-orders-subP-2 is there with partner: acme. Pub/Sub adds the source, partner-orders-sub, and the delivery count: 5. In production, the Pub/Sub service agent needs Publisher on the dead-letter topic and Subscriber on the source subscription.
If the project sits in a VPC Service Controls perimeter, the partner needs an ingress rule. Publishing is billed to the topic's project, and quota to the project of the caller's credentials.
Question 6: Pub/Sub costs jumped this month. Where do you look first?
First, the billing report, grouped by SKU, to see which part jumped: throughput, storage, or data transfer. Then the Pub/Sub metrics, like topic/send_request_count and byte cost, to find which topic or subscription.
| Bill line | Usual cause | Fix |
|---|---|---|
| Throughput (per TiB) | tiny messages, one per request: 1 KB minimum each | batch on the publisher |
| Throughput (per TiB) | filtered-out messages still pay delivery | a separate topic |
| Storage (per GiB-month) | retention, retained acks, snapshots, backlog over a day | right-size retention |
| Data transfer | subscribers in another region | read in the region |
| Export, SMT, import (per TiB) | BigQuery and Cloud Storage subscriptions | budget: no free tier |
Throughput is billed per tebibyte, for publishing and for every subscription's delivery, with a 1 KB minimum per request. So tiny messages sent one per request cost more than their size: batch on the publisher.
Every subscription pays its own delivery, and messages a filter drops still count as delivery throughput. If a subscriber filters out almost everything, a separate topic is cheaper.
Storage is the quiet one: topic retention, retained acked messages, snapshots, and unacked backlog older than a day.
Then data transfer, from subscribers in another region than the publishers. And export subscriptions, transforms and import topics are billed per tebibyte too, outside the monthly free tier.
A forgotten retain acked messages on a busy subscription adds up.
Scenario 7: In the final round, the hiring manager asks you to design an event platform on Google Cloud for an online retailer. What is your design?
I start from the domains: orders, payments, inventory and clicks, one topic each, with an Avro schema as the contract, and attributes like event type for routing.
Then one subscription per consumer. Let's put two on the orders topic: a refunds service filtered on event type, and analytics with no filter. Then six events go in. We run python subs.py, python events.py, and python read.py for each subscription.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
subscriber = pubsub_v1.SubscriberClient()
topic = f"projects/{project}/topics/orders"
for name, flt in [("refunds-svc", 'attributes.event_type = "refund"'),
("analytics", "")]:
s = subscriber.create_subscription(request={
"name": subscriber.subscription_path(project, name),
"topic": topic, "filter": flt})
print(name, "| filter:", s.filter or "none")from google.cloud import pubsub_v1
events = [("o-501", "order_created"), ("o-501", "payment"),
("o-502", "order_created"), ("o-501", "refund"),
("o-502", "payment"), ("o-503", "refund")]
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "orders")
for order, kind in events:
publisher.publish(topic, f"{order} {kind}".encode(),
event_type=kind).result()
print("published", len(events), "events to orders")import sys
from collections import Counter
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", sys.argv[1])
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
msgs = resp.received_messages
kinds = Counter(m.message.attributes["event_type"] for m in msgs)
print(sys.argv[1], "got", len(msgs), "|", dict(sorted(kinds.items())))
ack_ids = [m.ack_id for m in msgs]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)$ python subs.py
refunds-svc | filter: attributes.event_type = "refund"
analytics | filter: none
$ python events.py
published 6 events to orders
$ python read.py refunds-svc
refunds-svc got 2 | {'refund': 2}
$ python read.py analytics
analytics got 6 | {'order_created': 2, 'payment': 2, 'refund': 2}refunds-svc got 2 | {'refund': 2}, and analytics got 6. Each consumer gets exactly what it needs from one topic.
Then delivery per need:
| Consumer | Subscription |
|---|---|
| small services on Cloud Run | push, with OIDC |
| heavy workers | StreamingPull |
| analytics | BigQuery subscription |
| raw archive | Cloud Storage subscription |
| windows and joins | Dataflow |
Ordering keys and exactly-once only where needed, like inventory per product: ordering is 1 MB per second per key, and exactly-once adds latency. Every subscription gets a dead-letter topic and a retry policy, and topics keep retention for replay.
Around it:
- Ordering keys + exactly-once - only where needed, like inventory per product
- Dead-letter topic + retry policy - on every subscription
- Topic retention - for replay with seek
- Least-privilege IAM, CMEK, storage policy - per topic and subscription
- Alerts -
_by_regionbacklog metrics - Managed Service for Apache Kafka - only for the Kafka API
CMEK and a storage policy go in where required, and Managed Service for Apache Kafka only for teams that need the Kafka API.
Key points to remember
- Backlog: _by_region metrics, then scale subscribers
- Bad acks: seek, with retained acks or topic retention; snapshot before deploys
- Least-privilege IAM per topic and subscription
- Two regions: clients in both, failover in the runbook
- Partners: own topic, schema, dead-letter topic; costs: SKU first
The full script
This is a Python script that writes every file and runs every command above in order against the Pub/Sub emulator: start the emulator, run pip install google-cloud-pubsub first, and run it in an empty folder.
Show the full script that writes the files and runs every step
# ========================================================================
# Pub/Sub Interview Q&A #10 - Real-world manager scenarios
# 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_10_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"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path(project, "orders")
publisher.create_topic(name=topic)
subscriber.create_subscription(
name=subscriber.subscription_path(project, "orders-fulfilment"),
topic=topic)
""")
write('backlog.txt', """\
0
""")
run('python setup.py')
# Step 1: Evening peak: 30 orders in per tick, one worker acks 10
write('evening.py', """\
import sys
from google.cloud import pubsub_v1
incoming, workers, ticks = (int(a) for a in sys.argv[1:])
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path("tb-shop-dev", "orders")
sub = subscriber.subscription_path("tb-shop-dev", "orders-fulfilment")
backlog = int(open("backlog.txt").read()) # published minus acked
for tick in range(1, ticks + 1):
for i in range(incoming): # the website publishes
publisher.publish(topic, b'{"order": %d}' % i).result()
acked = 0
for w in range(workers): # 10 per worker
resp = subscriber.pull(subscription=sub, max_messages=10,
timeout=10)
ack_ids = [m.ack_id for m in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
acked += len(ack_ids)
backlog += incoming - acked
print(f"tick {tick}: in {incoming}, acked {acked},",
f"backlog {backlog}")
open("backlog.txt", "w").write(str(backlog))
""")
run('python evening.py 30 1 5')
# Step 2: The fix: six workers share the same subscription and drain the backlog
write('check.py', """\
from google.api_core.exceptions import DeadlineExceeded
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "orders-fulfilment")
try:
resp = subscriber.pull(subscription=sub, max_messages=10,
timeout=5)
left = len(resp.received_messages)
except DeadlineExceeded: # nothing to deliver
left = 0
print("messages still waiting in orders-fulfilment:", left)
""")
run('python evening.py 10 6 2')
run('python check.py')
# Step 3: payments-ledger retains acked messages; consumer v1 processes three payments
write('create_ledger.py', """\
from google.cloud import pubsub_v1
from google.protobuf import duration_pb2
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
sub = subscriber.subscription_path("tb-shop-dev", "payments-ledger")
publisher.create_topic(name=topic)
s = subscriber.create_subscription(request={
"name": sub, "topic": topic,
"retain_acked_messages": True, # keep acked messages
"message_retention_duration": duration_pb2.Duration(
seconds=7 * 24 * 3600)}) # for 7 days
print("retain_acked_messages:", s.retain_acked_messages,
"| retention:", s.message_retention_duration)
""")
write('pay.py', """\
import sys
from google.cloud import pubsub_v1
first, last = int(sys.argv[1]), int(sys.argv[2])
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "payments")
names = [f"pay-{i}" for i in range(first, last + 1)]
for name in names:
publisher.publish(topic, name.encode()).result()
print("published", ", ".join(names))
""")
write('consumer.py', """\
import sys
from google.cloud import pubsub_v1
version = sys.argv[1]
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "payments-ledger")
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
processed = []
for m in resp.received_messages:
if version != "v2": # v2 bug: acks, never processes
processed.append(m.message.data.decode())
ack_ids = [m.ack_id for m in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
print(version, "acked", len(ack_ids),
"| processed", sorted(processed))
""")
run('python create_ledger.py')
run('python pay.py 1 3')
run('python consumer.py v1')
# Step 4: Consumer v2 is released: it acks five payments without processing them
write('release.py', """\
import time
from datetime import datetime, timezone
time.sleep(1.5) # margin before the release
with open("release.txt", "w") as f:
f.write(datetime.now(timezone.utc).isoformat())
time.sleep(1.5)
print("consumer v2 released, release time saved")
""")
run('python release.py')
run('python pay.py 4 8')
run('python consumer.py v2')
# Step 5: Seek back to the release time and reprocess with the fixed consumer v3
write('seek.py', """\
from datetime import datetime
from google.cloud import pubsub_v1
from google.protobuf import timestamp_pb2
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", "payments-ledger")
release = datetime.fromisoformat(open("release.txt").read())
ts = timestamp_pb2.Timestamp()
ts.FromDatetime(release)
subscriber.seek(request={"subscription": sub, "time": ts})
print("payments-ledger rewound to the v2 release time")
""")
run('python seek.py')
run('python consumer.py v3')
# Step 6: A partner ingress topic with a dead-letter topic after 5 attempts
write('create_partner.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic = publisher.topic_path(project, "partner-orders")
dlq = publisher.topic_path(project, "partner-orders-dlq")
for t in (topic, dlq):
publisher.create_topic(name=t)
sub_path = subscriber.subscription_path
subscriber.create_subscription( # so dead letters are kept
name=sub_path(project, "partner-orders-dlq-sub"), topic=dlq)
s = subscriber.create_subscription(request={
"name": sub_path(project, "partner-orders-sub"),
"topic": topic,
"dead_letter_policy": {"dead_letter_topic": dlq,
"max_delivery_attempts": 5}})
p = s.dead_letter_policy
print("dead letter topic:", p.dead_letter_topic)
print("max delivery attempts:", p.max_delivery_attempts)
""")
run('python create_partner.py')
# Step 7: The partner sends three orders, one is broken JSON
write('partner_send.py', """\
from google.cloud import pubsub_v1
orders = ['{"order": "P-1", "amount": 40}',
'{"order": "P-2", "amount": ', # broken JSON
'{"order": "P-3", "amount": 15}']
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "partner-orders")
for o in orders:
publisher.publish(topic, o.encode(), partner="acme").result()
print("acme sent", len(orders), "orders")
""")
write('partner_consumer.py', """\
import json
from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = "projects/tb-shop-dev/subscriptions/partner-orders-sub"
done, attempts = [], []
def callback(m):
try:
done.append(json.loads(m.data)["order"])
m.ack()
except ValueError: # malformed: give it back
attempts.append(m.delivery_attempt)
m.nack()
future = subscriber.subscribe(sub, callback=callback)
try:
future.result(timeout=8)
except TimeoutError:
future.cancel()
future.result()
print("processed", sorted(done))
print("broken JSON nacked, delivery attempts", sorted(attempts))
""")
write('read_dlq.py', """\
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = "projects/tb-shop-dev/subscriptions/partner-orders-dlq-sub"
resp = subscriber.pull(subscription=sub, max_messages=10, timeout=10)
for r in resp.received_messages:
a = r.message.attributes
print("dead letter:", r.message.data.decode())
print(" partner:", a["partner"], "| delivery count:",
a["CloudPubSubDeadLetterSourceDeliveryCount"])
print(" source:", a["CloudPubSubDeadLetterSourceSubscription"])
ack_ids = [r.ack_id for r in resp.received_messages]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
""")
run('python partner_send.py')
run('python partner_consumer.py')
# Step 8: The broken order is in the dead-letter topic
run('python read_dlq.py')
# Step 9: One orders topic: a filtered refunds subscription and an analytics subscription
write('subs.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
subscriber = pubsub_v1.SubscriberClient()
topic = f"projects/{project}/topics/orders"
for name, flt in [("refunds-svc", 'attributes.event_type = "refund"'),
("analytics", "")]:
s = subscriber.create_subscription(request={
"name": subscriber.subscription_path(project, name),
"topic": topic, "filter": flt})
print(name, "| filter:", s.filter or "none")
""")
write('events.py', """\
from google.cloud import pubsub_v1
events = [("o-501", "order_created"), ("o-501", "payment"),
("o-502", "order_created"), ("o-501", "refund"),
("o-502", "payment"), ("o-503", "refund")]
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path("tb-shop-dev", "orders")
for order, kind in events:
publisher.publish(topic, f"{order} {kind}".encode(),
event_type=kind).result()
print("published", len(events), "events to orders")
""")
write('read.py', """\
import sys
from collections import Counter
from google.cloud import pubsub_v1
subscriber = pubsub_v1.SubscriberClient()
sub = subscriber.subscription_path("tb-shop-dev", sys.argv[1])
resp = subscriber.pull(subscription=sub, max_messages=50, timeout=10)
msgs = resp.received_messages
kinds = Counter(m.message.attributes["event_type"] for m in msgs)
print(sys.argv[1], "got", len(msgs), "|", dict(sorted(kinds.items())))
ack_ids = [m.ack_id for m in msgs]
subscriber.acknowledge(subscription=sub, ack_ids=ack_ids)
""")
run('python subs.py')
run('python events.py')
run('python read.py refunds-svc')
run('python read.py analytics')
All 10 parts are listed on the Real Time Pub/Sub Interview Questions and Answers page on this blog.
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.