In this post we are going to cover dead-letter topics and retries 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 partner sends malformed events every night and the pipeline stalls.
Every streaming pipeline gets bad messages sooner or later. A dead-letter topic is where Pub/Sub sends a message that failed too many times, and a retry policy controls how fast a failed message comes back. Interviewers ask about this because one poison message can stall a subscription. A good answer shows when to retry, when to give up, and how to get the data back after the fix.
Part 6 of the Real Time Pub/Sub Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: One bad message is redelivered forever and blocks progress. How do dead-letter topics help?
- Question 2: A teammate asks how Pub/Sub decides a message has failed too many times. How do delivery attempts work?
- Question 3: Retries hammer a database that is already down. How does the retry policy with backoff help?
- Question 4: The dead-letter topic fills up and nobody notices. How do you monitor and replay dead letters?
- Question 5: Dead lettering is set up but messages never move. Which permissions does Pub/Sub need?
- Question 6: Your manager asks when to retry and when to give up. How do you classify errors?
- Scenario 7: A partner sends malformed events every night and the pipeline stalls. How do you make it resilient?
- The full script
The short answer
A dead-letter policy caps retries: after max_delivery_attempts (5 to 100, default 5) Pub/Sub forwards the message to a dead-letter topic. delivery_attempt counts nacks and expired ack deadlines, and a retry policy adds exponential backoff. Grant the Pub/Sub service agent publisher and subscriber, alert on dead letters, replay after the fix, and park permanent errors instead of retrying them.
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, like IAM, are shown as code. We run python setup.py once before the first step. It creates the topic partner-events, the dead-letter topic partner-events-dlq with its own subscription, and the subscription partner-events-sub with a dead-letter policy of 5 attempts. publish.py publishes a partner file line by line with the attribute partner=acme, and consumer.py nacks any amount it can't parse.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "partner-events")
dlq = pub.topic_path(project, "partner-events-dlq")
for t in [topic, dlq]:
pub.create_topic(name=t)
sub.create_subscription(request={ # someone must read the DLQ
"name": sub.subscription_path(project, "partner-events-dlq-sub"),
"topic": dlq,
})
sub.create_subscription(request={
"name": sub.subscription_path(project, "partner-events-sub"),
"topic": topic,
"dead_letter_policy": {"dead_letter_topic": dlq,
"max_delivery_attempts": 5},
})import sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "partner-events")
lines = open(sys.argv[1]).read().splitlines()
for line in lines:
pub.publish(topic, line.encode(), partner="acme").result()
print("published", len(lines), "partner events from", sys.argv[1]){"order_id": 1, "amount": "25.00"}
{"order_id": 2, "amount": "12,50"}
{"order_id": 3, "amount": "9.99"}import json, threading
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
seen, done, lock = {}, {}, threading.Lock()
def callback(message):
event = json.loads(message.data)
order = event["order_id"]
with lock:
seen.setdefault(order, []).append(message.delivery_attempt)
try:
amount = float(event["amount"])
except ValueError:
message.nack() # can't parse it: give it back
return
done[order] = amount
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=8)
except TimeoutError:
future.cancel()
future.result()
for order in sorted(seen):
result = "processed" if order in done else "nacked"
print("order", order, result, "on delivery attempts", seen[order])from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "partner-events-dlq-sub")
resp = sub.pull(subscription=path, max_messages=10, timeout=30)
for r in resp.received_messages:
print("dead letter:", r.message.data.decode())
for key, value in sorted(r.message.attributes.items()):
if key.endswith("PublishTime"):
value = "(the publish time)"
print(" ", key, "=", value)
ack_ids = [r.ack_id for r in resp.received_messages]
sub.modify_ack_deadline(subscription=path, ack_ids=ack_ids,
ack_deadline_seconds=0) # keep it for replayQuestion 1: One bad message is redelivered forever and blocks progress. How do dead-letter topics help?
A dead-letter topic caps the retries. We set a dead-letter policy on the subscription: a topic, and max delivery attempts, from 5 to 100, 5 by default. After that many failed deliveries, Pub/Sub forwards the message to that topic and removes it from the subscription.
Without it, a poison message comes back forever:
- It wastes consumer capacity
- The oldest unacked age keeps growing
- On an ordered subscription, it blocks its key
Let's try it. The partner file has three orders, and order 2 has the amount "12,50", with a comma. Our consumer can't parse it, so it nacks. We run python publish.py acme_orders.jsonl, then python consumer.py.
$ python publish.py acme_orders.jsonl
published 3 partner events from acme_orders.jsonl
$ python consumer.py
order 1 processed on delivery attempts [1]
order 2 nacked on delivery attempts [1, 2, 3, 4, 5]
order 3 processed on delivery attempts [1]Orders 1 and 3 are processed on the first attempt, [1]. Order 2 is nacked on delivery attempts [1, 2, 3, 4, 5]. Now let's read the dead-letter subscription with python read_dlq.py.
$ python read_dlq.py
dead letter: {"order_id": 2, "amount": "12,50"}
CloudPubSubDeadLetterSourceDeliveryCount = 5
CloudPubSubDeadLetterSourceSubscription = partner-events-sub
CloudPubSubDeadLetterSourceSubscriptionProject = tb-shop-dev
CloudPubSubDeadLetterSourceTopicPublishTime = (the publish time)
partner = acmeOrder 2 is there, {"order_id": 2, "amount": "12,50"}, with its original data and its attribute partner = acme. That is why setup.py creates the dead-letter subscription together with the topic: messages forwarded to a topic with no subscription are lost.
Forwarding is best effort, so a message can occasionally be dead-lettered more than once. Make the dead-letter readers idempotent too.
Question 2: A teammate asks how Pub/Sub decides a message has failed too many times. How do delivery attempts work?
Pub/Sub counts delivery attempts per message: the first delivery, plus every nack, plus every ack deadline that expired. Once the maximum number of attempts has failed, the message is dead-lettered.
The subscriber sees the count: delivery_attempt in pull responses, and deliveryAttempt in the push request body. It is only set when the subscription has a dead-letter policy. Otherwise it is 0 or empty.
Let's check each part. attempts.py publishes order 4 and pulls it. Then it nacks it, by setting the ack deadline to 0, and pulls again. Then it holds the message past its 10 second ack deadline and pulls a third time. We run python attempts.py.
import time
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "partner-events")
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
pub.publish(topic, b'{"order_id": 4, "amount": "5.00"}').result()
def pull(label):
resp = sub.pull(subscription=path, max_messages=1, timeout=30)
m = resp.received_messages[0]
print(label, "-> delivery_attempt", m.delivery_attempt)
return m.ack_id
ack_id = pull("first pull")
sub.modify_ack_deadline(subscription=path, ack_ids=[ack_id],
ack_deadline_seconds=0) # a nack
ack_id = pull("after a nack")
time.sleep(13) # the 10 s ack deadline expires
ack_id = pull("after the ack deadline expired")
sub.acknowledge(subscription=path, ack_ids=[ack_id])$ python attempts.py
first pull -> delivery_attempt 1
after a nack -> delivery_attempt 2
after the ack deadline expired -> delivery_attempt 3The first pull shows delivery_attempt 1, after a nack it is 2, and after the ack deadline expired it is 3. So letting a deadline expire burns attempts too, not just errors.
The count is approximate and best effort, and seeking the subscription resets it to 0. The policy itself is validated. Let's ask for 4 attempts with python bad_policy.py.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
sub = pubsub_v1.SubscriberClient()
dlq = f"projects/{project}/topics/partner-events-dlq"
try:
sub.create_subscription(request={
"name": sub.subscription_path(project, "partner-events-fast"),
"topic": f"projects/{project}/topics/partner-events",
"dead_letter_policy": {"dead_letter_topic": dlq,
"max_delivery_attempts": 4},
}, timeout=30)
except Exception as e:
print(type(e).__name__, e)$ python bad_policy.py
OutOfRange 400 The value for max_delivery_attempts is too small. You passed 4 in the request, but the
minimum value is 5.We get OutOfRange 400 The value for max_delivery_attempts is too small. You passed 4 in the request, but the minimum value is 5.
Log the delivery attempt with every failure. A message on attempt 4 is your early warning, before it reaches the dead-letter topic.
Question 3: Retries hammer a database that is already down. How does the retry policy with backoff help?
Without a retry policy, a nacked message is redelivered right away, so a database that is down gets hammered. A retry policy on the subscription adds exponential backoff between a minimum and a maximum delay, each 0 to 600 seconds.
Let's create three subscriptions with python retry.py: the orders DB writer with 10 seconds minimum and 300 maximum, one with an empty policy, and one with a maximum of 601 seconds.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
sub = pubsub_v1.SubscriberClient()
topic = f"projects/{project}/topics/partner-events"
def create(name, policy):
try:
s = sub.create_subscription(request={
"name": sub.subscription_path(project, name),
"topic": topic, "retry_policy": policy}, timeout=30)
low = s.retry_policy.minimum_backoff.total_seconds()
high = s.retry_policy.maximum_backoff.total_seconds()
print(name, "-> backoff", int(low), "s to", int(high), "s")
except Exception as e:
print(name, "->", type(e).__name__, e)
create("orders-db-writer", {"minimum_backoff": {"seconds": 10},
"maximum_backoff": {"seconds": 300}})
create("orders-defaults", {})
create("orders-too-slow", {"maximum_backoff": {"seconds": 601}})$ python retry.py
orders-db-writer -> backoff 10 s to 300 s
orders-defaults -> backoff 10 s to 600 s
orders-too-slow -> InvalidArgument 400 The value for maximum_backoff is out of bounds. You passed 601s
in the request, but the value must be between 0s and 600s.The writer gets backoff 10 s to 300 s. The empty policy takes the defaults, 10 s to 600 s. And 601 is rejected: You passed 601s in the request, but the value must be between 0s and 600s. We don't time the backoff here. The emulator stores and validates the policy, but its timing is loose.
Here is how the backoff behaves:
- Triggers - nacks and expired ack deadlines, best effort
- Batches - backoff starts once every message in the batch is nacked or expired
- Push - its own push backoff too: the larger delay wins
The trap is retries plus a dead-letter policy. If the database is down for an hour, five attempts go quickly, and good messages end up in the dead-letter topic. During a real outage, pause the consumer instead of nacking everything.
The attempts times the backoff should outlast a normal blip. And alert on the backlog, so a real outage pages someone.
Question 4: The dead-letter topic fills up and nobody notices. How do you monitor and replay dead letters?
Alert on it. The source subscription has a dead-letter message count metric, and the dead-letter subscription has its own backlog and oldest unacked age.
| Metric (pubsub.googleapis.com/) | Watch it on | Alert when |
|---|---|---|
| subscription/dead_letter_message_count | the source subscription | above 0 |
| subscription/num_unacked_messages_by_region | the dead-letter subscription | above 0 |
| subscription/oldest_unacked_message_age_by_region | the dead-letter subscription | growing |
Anything above zero should open a ticket with an owner.
Each dead letter tells us where it came from. Let's go back to the one we read in question 1. Pub/Sub keeps the original data and attributes, partner = acme, and adds the source subscription partner-events-sub, its project tb-shop-dev, the original publish time, and CloudPubSubDeadLetterSourceDeliveryCount = 5. read_dlq.py doesn't ack it, so it stays there for the replay.
Replay is a small tool we own. Pull from the dead-letter subscription, republish to the main topic with the original attributes, and ack the dead letter only after the publish succeeded.
But fix first. We deploy a consumer that accepts the partner's decimal comma, with amount.replace(",", "."). replay.py drops the CloudPubSubDeadLetter attributes and adds replayed="true". We run python replay.py, then python consumer.py.
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "partner-events")
dlq_sub = sub.subscription_path(project, "partner-events-dlq-sub")
resp = sub.pull(subscription=dlq_sub, max_messages=10, timeout=30)
for r in resp.received_messages:
attrs = {k: v for k, v in r.message.attributes.items()
if not k.startswith("CloudPubSubDeadLetter")}
pub.publish(topic, r.message.data, **attrs,
replayed="true").result()
sub.acknowledge(subscription=dlq_sub, ack_ids=[r.ack_id]) # after
print("replayed", r.message.data.decode(), attrs)import json, threading
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
seen, done, lock = {}, {}, threading.Lock()
def callback(message):
event = json.loads(message.data)
order = event["order_id"]
with lock:
seen.setdefault(order, []).append(message.delivery_attempt)
try:
amount = float(event["amount"].replace(",", ".")) # the fix
except ValueError:
message.nack() # can't parse it: give it back
return
done[order] = amount
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=8)
except TimeoutError:
future.cancel()
future.result()
for order in sorted(seen):
result = "processed" if order in done else "nacked"
print("order", order, result, "on delivery attempts", seen[order])$ python replay.py
replayed {"order_id": 2, "amount": "12,50"} {'partner': 'acme'}
$ python consumer.py
order 2 processed on delivery attempts [1]The replay prints replayed {"order_id": 2, "amount": "12,50"} {'partner': 'acme'}, and then order 2 processed on delivery attempts [1]. A replayed message is a new message.
Replay in small batches, with an audit attribute like replayed=true. Republishing reaches every subscription on that topic, so those consumers must be idempotent too. The dead-letter subscription keeps messages 7 days by default.
Question 5: Dead lettering is set up but messages never move. Which permissions does Pub/Sub need?
Pub/Sub forwards dead letters as its own service agent: service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com. That identity needs two grants:
roles/pubsub.publisheron the dead-letter topicpartner-events-dlq- so it can forward the messageroles/pubsub.subscriberon the source subscriptionpartner-events-sub- so it can acknowledge the forwarded message and remove it
With gcloud, we look up the project number, build the service agent's email, and add the two bindings. The emulator has IAM turned off, so this part is code only.
gcloud config set project tb-shop-prod
PROJECT_NUMBER=$(gcloud projects describe tb-shop-prod \
--format="value(projectNumber)")
SA="service-${PROJECT_NUMBER}"
SA="${SA}@gcp-sa-pubsub.iam.gserviceaccount.com"
gcloud pubsub topics add-iam-policy-binding partner-events-dlq \
--member="serviceAccount:${SA}" --role="roles/pubsub.publisher"
gcloud pubsub subscriptions add-iam-policy-binding \
partner-events-sub \
--member="serviceAccount:${SA}" --role="roles/pubsub.subscriber"If messages never move, check these, in order:
- Both grants - on the right topic and subscription; without them, attempts aren't counted
- The dead-letter topic exists - and has a subscription
- Attempts reach the maximum - a consumer that acks everything never dead letters
With Terraform or gcloud, the grants must be in your code, not added by hand.
Question 6: Your manager asks when to retry and when to give up. How do you classify errors?
Ask one question: will trying again help? Then each kind of error gets its own action.
| Error | Examples | Action |
|---|---|---|
| Transient | timeout, unavailable, a database that is down | nack: the retry policy backs off |
| Permanent | bad JSON, a missing field | park it with the reason, ack |
| Needs a person | permission denied | stop and alert |
| Unexpected | a bug nobody planned for | nack: the dead-letter policy is the safety net |
Transient errors, like a timeout or a database that is down, will pass: nack, and let the retry policy back off. Permanent errors, like bad JSON or a missing field, fail every time. So don't spend five attempts on them: park the raw message with the reason, and ack it. Errors that need a person, like permission denied, aren't the message's fault: stop and alert. For anything unexpected, nack, and let the dead-letter policy be the safety net.
Let's try it. db.py is our stand-in for the warehouse write, and order 7 times out once. classify.py parks bad data in the DLQ topic with the error as an attribute. We run python publish.py acme_night2.jsonl, python classify.py and python read_dlq.py.
calls = {}
def write_to_db(event):
"""Stand-in for the warehouse write: order 7 times out once."""
order = event["order_id"]
calls[order] = calls.get(order, 0) + 1
if order == 7 and calls[order] == 1:
raise TimeoutError("database timeout")import json, time
from google.cloud import pubsub_v1
from db import write_to_db
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
dlq = pub.topic_path("tb-shop-dev", "partner-events-dlq")
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
log = []
def callback(message):
try:
write_to_db(json.loads(message.data))
action = "written, acked"
message.ack()
except (ValueError, KeyError) as e: # permanent: bad data
pub.publish(dlq, message.data, error=str(e)).result()
action = "parked in the DLQ with the error, acked"
message.ack()
except TimeoutError: # transient: try again
action = "nacked, retried later"
message.nack()
log.append((message.data, message.delivery_attempt, action))
future = sub.subscribe(path, callback=callback)
time.sleep(8)
future.cancel()
future.result()
for data, attempt, action in sorted(log):
print(f"{data.decode():16} attempt {attempt}: {action}"){"order_id": 5}
not json
{"order_id": 7}$ python publish.py acme_night2.jsonl
published 3 partner events from acme_night2.jsonl
$ python classify.py
not json attempt 1: parked in the DLQ with the error, acked
{"order_id": 5} attempt 1: written, acked
{"order_id": 7} attempt 1: nacked, retried later
{"order_id": 7} attempt 2: written, acked
$ python read_dlq.py
dead letter: not json
error = Expecting value: line 1 column 1 (char 0)not json is parked on attempt 1 and acked. Order 5 is written on attempt 1. Order 7 is nacked, retried later on attempt 1, and written on attempt 2. In the DLQ the message carries its reason: error = Expecting value: line 1 column 1 (char 0).
Two details. The consumer needs publisher on the parking topic. And the reason must travel with the message, because Pub/Sub adds a delivery error message attribute only for export subscriptions.
Keep the classification in one place, test it with real bad messages, and count each class, so a spike in bad data goes to the team that owns it.
Scenario 7: A partner sends malformed events every night and the pipeline stalls. How do you make it resilient?
We build it in layers. First, stop bad data at the door with a schema on the topic. schema_topic.py creates the Avro schema partner-order and the topic partner-orders with JSON encoding, then publishes two orders. We run python schema_topic.py.
from google.cloud import pubsub_v1
from google.pubsub_v1.types import Encoding, Schema
project = "tb-shop-dev"
schemas = pubsub_v1.SchemaServiceClient()
pub = pubsub_v1.PublisherClient()
avro = """{"type": "record", "name": "PartnerOrder", "fields": [
{"name": "order_id", "type": "int"},
{"name": "amount", "type": "double"}]}"""
schema = schemas.create_schema(parent=f"projects/{project}",
schema=Schema(type_=Schema.Type.AVRO, definition=avro),
schema_id="partner-order")
topic = pub.topic_path(project, "partner-orders")
pub.create_topic(request={"name": topic, "schema_settings": {
"schema": schema.name, "encoding": Encoding.JSON}})
for data in ['{"order_id": 8, "amount": 19.99}',
'{"order_id": 9, "amount": "12,50"}']:
try:
pub.publish(topic, data.encode()).result(timeout=30)
print("accepted", data)
except Exception as e:
print("rejected", data)
print(" ", type(e).__name__, e)$ python schema_topic.py
accepted {"order_id": 8, "amount": 19.99}
rejected {"order_id": 9, "amount": "12,50"}
InvalidArgument 400 Could not parse messageOrder 8 is accepted. Order 9, with the amount as a string, is rejected with InvalidArgument 400 Could not parse message. It fails at publish, so the partner sees the error, not our pipeline.
Second, no subscription retries forever: a dead-letter topic with 5 to 10 attempts, a retry policy with backoff, and a subscription on the dead-letter topic. With gcloud it looks like this.
gcloud pubsub subscriptions create partner-orders-sub \
--topic=partner-orders \
--dead-letter-topic=partner-orders-dlq \
--max-delivery-attempts=10 \
--min-retry-delay=10s --max-retry-delay=600sThird, the consumer classifies, like in question 6. Bad data is parked with the reason and acked, and a blip is nacked and retried. So poison messages can't keep coming back, and good events keep moving. Fourth, alert on the dead-letter count, and replay with a tool after the fix. Replayed events go through the same checks.
Here is the whole design:
- Schema on the topic - bad events rejected at publish
- Dead-letter topic + retry policy - 5 to 10 attempts, backoff, a DLQ subscription
- Classify in the consumer - park bad data, retry blips
- Alert and replay - dead_letter_message_count, a replay tool
- Schema = the contract - rejected events back to the partner every morning
Share the schema with the partner as the contract, send them the rejected events each morning, and agree on fixes. The pipeline stays green, and bad data has an owner.
Key points to remember
- Dead-letter policy: after 5 to 100 attempts, forward to a dead-letter topic
- delivery_attempt = first delivery + nacks + expired ack deadlines
- Retry policy: exponential backoff, 0 to 600 s, defaults 10 s and 600 s
- Service agent: publisher on the DLQ topic, subscriber on the source
- Park permanent errors, retry transient ones, alert and replay after the fix
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 #6 - Dead-letter topics and retries
# 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_06_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, "partner-events")
dlq = pub.topic_path(project, "partner-events-dlq")
for t in [topic, dlq]:
pub.create_topic(name=t)
sub.create_subscription(request={ # someone must read the DLQ
"name": sub.subscription_path(project, "partner-events-dlq-sub"),
"topic": dlq,
})
sub.create_subscription(request={
"name": sub.subscription_path(project, "partner-events-sub"),
"topic": topic,
"dead_letter_policy": {"dead_letter_topic": dlq,
"max_delivery_attempts": 5},
})
""")
write('publish.py', """\
import sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "partner-events")
lines = open(sys.argv[1]).read().splitlines()
for line in lines:
pub.publish(topic, line.encode(), partner="acme").result()
print("published", len(lines), "partner events from", sys.argv[1])
""")
write('acme_orders.jsonl', """\
{"order_id": 1, "amount": "25.00"}
{"order_id": 2, "amount": "12,50"}
{"order_id": 3, "amount": "9.99"}
""")
write('consumer.py', """\
import json, threading
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
seen, done, lock = {}, {}, threading.Lock()
def callback(message):
event = json.loads(message.data)
order = event["order_id"]
with lock:
seen.setdefault(order, []).append(message.delivery_attempt)
try:
amount = float(event["amount"])
except ValueError:
message.nack() # can't parse it: give it back
return
done[order] = amount
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=8)
except TimeoutError:
future.cancel()
future.result()
for order in sorted(seen):
result = "processed" if order in done else "nacked"
print("order", order, result, "on delivery attempts", seen[order])
""")
write('read_dlq.py', """\
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "partner-events-dlq-sub")
resp = sub.pull(subscription=path, max_messages=10, timeout=30)
for r in resp.received_messages:
print("dead letter:", r.message.data.decode())
for key, value in sorted(r.message.attributes.items()):
if key.endswith("PublishTime"):
value = "(the publish time)"
print(" ", key, "=", value)
ack_ids = [r.ack_id for r in resp.received_messages]
sub.modify_ack_deadline(subscription=path, ack_ids=ack_ids,
ack_deadline_seconds=0) # keep it for replay
""")
run('python setup.py')
# Step 1: A partner file with one bad amount: the consumer nacks it on every delivery attempt
run('python publish.py acme_orders.jsonl')
run('python consumer.py')
# Step 2: After 5 attempts the message is in the dead-letter subscription
run('python read_dlq.py')
# Step 3: delivery_attempt counts nacks and expired ack deadlines
write('attempts.py', """\
import time
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "partner-events")
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
pub.publish(topic, b'{"order_id": 4, "amount": "5.00"}').result()
def pull(label):
resp = sub.pull(subscription=path, max_messages=1, timeout=30)
m = resp.received_messages[0]
print(label, "-> delivery_attempt", m.delivery_attempt)
return m.ack_id
ack_id = pull("first pull")
sub.modify_ack_deadline(subscription=path, ack_ids=[ack_id],
ack_deadline_seconds=0) # a nack
ack_id = pull("after a nack")
time.sleep(13) # the 10 s ack deadline expires
ack_id = pull("after the ack deadline expired")
sub.acknowledge(subscription=path, ack_ids=[ack_id])
""")
run('python attempts.py')
# Step 4: max_delivery_attempts must be between 5 and 100
write('bad_policy.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
sub = pubsub_v1.SubscriberClient()
dlq = f"projects/{project}/topics/partner-events-dlq"
try:
sub.create_subscription(request={
"name": sub.subscription_path(project, "partner-events-fast"),
"topic": f"projects/{project}/topics/partner-events",
"dead_letter_policy": {"dead_letter_topic": dlq,
"max_delivery_attempts": 4},
}, timeout=30)
except Exception as e:
print(type(e).__name__, e)
""")
run('python bad_policy.py')
# Step 5: Retry policies: custom backoff, the defaults, and the bounds
write('retry.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
sub = pubsub_v1.SubscriberClient()
topic = f"projects/{project}/topics/partner-events"
def create(name, policy):
try:
s = sub.create_subscription(request={
"name": sub.subscription_path(project, name),
"topic": topic, "retry_policy": policy}, timeout=30)
low = s.retry_policy.minimum_backoff.total_seconds()
high = s.retry_policy.maximum_backoff.total_seconds()
print(name, "-> backoff", int(low), "s to", int(high), "s")
except Exception as e:
print(name, "->", type(e).__name__, e)
create("orders-db-writer", {"minimum_backoff": {"seconds": 10},
"maximum_backoff": {"seconds": 300}})
create("orders-defaults", {})
create("orders-too-slow", {"maximum_backoff": {"seconds": 601}})
""")
run('python retry.py')
# Step 6: Fix the consumer, replay the dead letter, process it
write('replay.py', """\
from google.cloud import pubsub_v1
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
topic = pub.topic_path(project, "partner-events")
dlq_sub = sub.subscription_path(project, "partner-events-dlq-sub")
resp = sub.pull(subscription=dlq_sub, max_messages=10, timeout=30)
for r in resp.received_messages:
attrs = {k: v for k, v in r.message.attributes.items()
if not k.startswith("CloudPubSubDeadLetter")}
pub.publish(topic, r.message.data, **attrs,
replayed="true").result()
sub.acknowledge(subscription=dlq_sub, ack_ids=[r.ack_id]) # after
print("replayed", r.message.data.decode(), attrs)
""")
write('consumer.py', """\
import json, threading
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
seen, done, lock = {}, {}, threading.Lock()
def callback(message):
event = json.loads(message.data)
order = event["order_id"]
with lock:
seen.setdefault(order, []).append(message.delivery_attempt)
try:
amount = float(event["amount"].replace(",", ".")) # the fix
except ValueError:
message.nack() # can't parse it: give it back
return
done[order] = amount
message.ack()
future = sub.subscribe(path, callback=callback)
try:
future.result(timeout=8)
except TimeoutError:
future.cancel()
future.result()
for order in sorted(seen):
result = "processed" if order in done else "nacked"
print("order", order, result, "on delivery attempts", seen[order])
""")
run('python replay.py')
run('python consumer.py')
# Step 7: Classify errors: park bad data with the reason, retry a transient failure
write('db.py', 'calls = {}\n\ndef write_to_db(event):\n """Stand-in for the warehouse write: order 7 times out once."""\n order = event["order_id"]\n calls[order] = calls.get(order, 0) + 1\n if order == 7 and calls[order] == 1:\n raise TimeoutError("database timeout")\n')
write('classify.py', """\
import json, time
from google.cloud import pubsub_v1
from db import write_to_db
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
dlq = pub.topic_path("tb-shop-dev", "partner-events-dlq")
path = sub.subscription_path("tb-shop-dev", "partner-events-sub")
log = []
def callback(message):
try:
write_to_db(json.loads(message.data))
action = "written, acked"
message.ack()
except (ValueError, KeyError) as e: # permanent: bad data
pub.publish(dlq, message.data, error=str(e)).result()
action = "parked in the DLQ with the error, acked"
message.ack()
except TimeoutError: # transient: try again
action = "nacked, retried later"
message.nack()
log.append((message.data, message.delivery_attempt, action))
future = sub.subscribe(path, callback=callback)
time.sleep(8)
future.cancel()
future.result()
for data, attempt, action in sorted(log):
print(f"{data.decode():16} attempt {attempt}: {action}")
""")
write('acme_night2.jsonl', """\
{"order_id": 5}
not json
{"order_id": 7}
""")
run('python publish.py acme_night2.jsonl')
run('python classify.py')
run('python read_dlq.py')
# Step 8: Scenario: a schema on the topic rejects malformed events at publish
write('schema_topic.py', 'from google.cloud import pubsub_v1\nfrom google.pubsub_v1.types import Encoding, Schema\n\nproject = "tb-shop-dev"\nschemas = pubsub_v1.SchemaServiceClient()\npub = pubsub_v1.PublisherClient()\navro = """{"type": "record", "name": "PartnerOrder", "fields": [\n {"name": "order_id", "type": "int"},\n {"name": "amount", "type": "double"}]}"""\nschema = schemas.create_schema(parent=f"projects/{project}",\n schema=Schema(type_=Schema.Type.AVRO, definition=avro),\n schema_id="partner-order")\ntopic = pub.topic_path(project, "partner-orders")\npub.create_topic(request={"name": topic, "schema_settings": {\n "schema": schema.name, "encoding": Encoding.JSON}})\nfor data in [\'{"order_id": 8, "amount": 19.99}\',\n \'{"order_id": 9, "amount": "12,50"}\']:\n try:\n pub.publish(topic, data.encode()).result(timeout=30)\n print("accepted", data)\n except Exception as e:\n print("rejected", data)\n print(" ", type(e).__name__, e)\n')
run('python schema_topic.py')
Part 7: Filtering, Replay and Retention
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.