In this post we are going to cover filtering, replay and retention 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 bad release corrupted a downstream table for three hours, and we recover the data with Pub/Sub.
A filter decides which messages a subscription gets. Replay means delivering messages again, with seek to a time or to a snapshot. Retention decides how long Pub/Sub keeps the messages, and so how far back we can replay. Interviewers ask about this because every team ships a bad release sooner or later, and the question is always the same: can we get the data back?
Part 7 of the Real Time Pub/Sub Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: A subscriber only needs refund events from a busy topic. How do subscription filters work?
- Question 2: A bug wrote wrong data for two hours. How do you replay messages with seek?
- Question 3: Someone asks how long Pub/Sub keeps messages. How do message retention and topic retention work?
- Question 4: A teammate wants to reprocess from a known point before a deploy. How do snapshots help?
- Question 5: Acked messages are needed again for a new consumer. What are the options?
- Question 6: A filter was added and old messages are still delivered. What are the rules for changing filters?
- Scenario 7: A bad release corrupted a downstream table for three hours. How do you recover the data with Pub/Sub?
- The full script
The short answer
A filter on a subscription delivers only messages whose attributes match; it can't be changed later. Seek replays a subscription from a time or a snapshot, as long as topic retention or retain acked messages still holds the data. Subscriptions keep unacked messages 7 days by default, up to 31 days.
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: filters, seek to a time, and snapshots. The emulator accepts retention settings but doesn't apply them, so those limits come from the documentation, and features it doesn't support are shown as code. We run python setup.py once before the first step. It creates three topics, payments, orders and sales, each with 7 days of topic retention, and one subscription on each: payments-all, orders-loader and sales-to-table. drain.py pulls a subscription until nothing is left and acks every batch.
from google.cloud import pubsub_v1
from google.protobuf.duration_pb2 import Duration
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
week = Duration(seconds=7 * 24 * 3600)
for topic, name in [("payments", "payments-all"),
("orders", "orders-loader"),
("sales", "sales-to-table")]:
path = pub.topic_path(project, topic)
pub.create_topic(request={"name": path,
"message_retention_duration": week})
sub.create_subscription(
name=sub.subscription_path(project, name), topic=path)from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
s = sub.create_subscription(request={
"name": sub.subscription_path("tb-shop-dev", "refunds-only"),
"topic": "projects/tb-shop-dev/topics/payments",
"filter": 'attributes.event_type = "refund"',
})
print("created", s.name.split("/")[-1], "with filter:", s.filter)import sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "payments")
kinds = ["payment", "refund", "payment",
"chargeback", "refund", "payment"]
for n, kind in enumerate(kinds, int(sys.argv[1])):
pub.publish(topic, f"txn-{n}".encode(), event_type=kind).result()
print("published", len(kinds), "events:", ", ".join(kinds))import sys
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", sys.argv[1])
got = []
while True: # pull until nothing is left, ack every batch
resp = sub.pull(subscription=path, max_messages=100, timeout=10)
if not resp.received_messages:
break
for m in resp.received_messages:
text = m.message.data.decode()
kind = m.message.attributes.get("event_type", "")
got.append(f"{text} {kind}".strip())
ids = [m.ack_id for m in resp.received_messages]
sub.acknowledge(subscription=path, ack_ids=ids)
print(sys.argv[1], "got", len(got), "|", ", ".join(sorted(got)))import sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "orders")
first, last = int(sys.argv[1]), int(sys.argv[2])
for n in range(first, last + 1):
pub.publish(topic, f"order-{n}".encode()).result()
print(f"published order-{first} to order-{last}")import datetime, time
time.sleep(1.5) # a clear gap before and after the mark
now = datetime.datetime.now(datetime.timezone.utc)
open("release_time.txt", "w").write(now.isoformat())
time.sleep(1.5)
print("release v2 deployed, time saved to release_time.txt")import datetime, sys
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", sys.argv[1])
if sys.argv[2] == "--time":
when = open("release_time.txt").read()
sub.seek(request={"subscription": path,
"time": datetime.datetime.fromisoformat(when)})
print("seek", sys.argv[1], "to the release time")
else:
snap = sub.snapshot_path("tb-shop-dev", sys.argv[2])
sub.seek(request={"subscription": path, "snapshot": snap})
print("seek", sys.argv[1], "to snapshot", sys.argv[2])from google.cloud import pubsub_v1
from google.protobuf.duration_pb2 import Duration
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
t = pub.get_topic(topic=pub.topic_path("tb-shop-dev", "orders"))
print("topic orders, retention:", t.message_retention_duration)
s = sub.get_subscription(subscription=sub.subscription_path(
"tb-shop-dev", "orders-loader"))
print("subscription orders-loader, retention:",
s.message_retention_duration)
print("retain acked messages:", s.retain_acked_messages)
for days in (31, 32):
try:
keep = Duration(seconds=days * 86400)
pub.create_topic(request={
"name": pub.topic_path("tb-shop-dev", f"clicks-{days}d"),
"message_retention_duration": keep})
print(days, "days of topic retention: accepted")
except Exception as e:
print(days, "days of topic retention:", type(e).__name__, e)import datetime, sys
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", sys.argv[1])
snap = sub.create_snapshot(request={
"name": sub.snapshot_path("tb-shop-dev", sys.argv[2]),
"subscription": path})
now = datetime.datetime.now(datetime.timezone.utc)
left = snap.expire_time - now
print("snapshot", sys.argv[2], "of topic", snap.topic.split("/")[-1])
print("expires in", round(left.total_seconds() / 86400), "days")import sys
from google.cloud import pubsub_v1
project = "tb-shop-dev"
name, topic, snapshot = sys.argv[1:4]
flt = sys.argv[4] if len(sys.argv) > 4 else ""
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path(project, name)
sub.create_subscription(request={
"name": path, "topic": f"projects/{project}/topics/{topic}",
"filter": flt})
sub.seek(request={"subscription": path,
"snapshot": sub.snapshot_path(project, snapshot)})
print("created", name, "filter:", repr(flt), "| seek to", snapshot)from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "payments-all")
refunds = 'attributes.event_type = "refund"'
try:
sub.update_subscription(request={
"subscription": {"name": path, "filter": refunds},
"update_mask": {"paths": ["filter"]}}, timeout=10, retry=None)
print("filter changed")
except Exception as e:
print(type(e).__name__, e)import json, sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "sales")
first, last = int(sys.argv[1]), int(sys.argv[2])
for n in range(first, last + 1):
event = {"order_id": n, "amount_cents": 1000 + n * 5}
pub.publish(topic, json.dumps(event).encode()).result()
print("published sales for orders", first, "to", last)import json, sqlite3, sys
from google.cloud import pubsub_v1
release = sys.argv[1]
db = sqlite3.connect("warehouse.db")
db.execute("CREATE TABLE IF NOT EXISTS sales"
" (order_id INTEGER PRIMARY KEY, amount REAL)")
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "sales-to-table")
UPSERT = ("INSERT INTO sales VALUES (?, ?) ON CONFLICT(order_id)"
" DO UPDATE SET amount = excluded.amount")
n = 0
while True:
resp = sub.pull(subscription=path, max_messages=100, timeout=10)
if not resp.received_messages:
break
for m in resp.received_messages:
e = json.loads(m.message.data)
cents = e["amount_cents"]
amount = cents if release == "v2" else cents / 100 # v2 bug
db.execute(UPSERT, (e["order_id"], amount))
n += 1
ids = [m.ack_id for m in resp.received_messages]
db.commit() # write first, then ack
sub.acknowledge(subscription=path, ack_ids=ids)
print(release, "loaded", n, "events")
for row in db.execute("SELECT * FROM sales ORDER BY order_id"):
print(row)Question 1: A subscriber only needs refund events from a busy topic. How do subscription filters work?
We set a filter when we create the subscription, and it matches message attributes, not the payload. Let's create refunds-only with the filter attributes.event_type = "refund", publish six payment events, each with an event_type attribute, and drain both subscriptions. We run python refunds.py, python publish_payments.py 101, then python drain.py for payments-all and for refunds-only.
$ python refunds.py
created refunds-only with filter: attributes.event_type = "refund"
$ python publish_payments.py 101
published 6 events: payment, refund, payment, chargeback, refund, payment
$ python drain.py payments-all
payments-all got 6 | txn-101 payment, txn-102 refund, txn-103 payment, txn-104 chargeback, txn-105
refund, txn-106 payment
$ python drain.py refunds-only
refunds-only got 2 | txn-102 refund, txn-105 refundThe unfiltered subscription prints payments-all got 6. The filtered one prints refunds-only got 2 | txn-102 refund, txn-105 refund.
The syntax is small:
=and!=-attributes.event_type = "refund":- attribute exists:attributes:event_typeAND,OR,NOT- uppercase; parentheses to mix AND and ORhasPrefix()- the only function:hasPrefix(attributes.event_type, "ref")- Not supported - the message data, regex, numeric compare; max 256 bytes
Messages that don't match are acknowledged automatically. We don't pay egress for them, but delivery throughput is still charged, and backlog metrics count only matching messages.
Ask publishers to put the event type in an attribute. To filter on the payload, use a single message transform with a JavaScript UDF instead.
Question 2: A bug wrote wrong data for two hours. How do you replay messages with seek?
Seek the subscription to a time just before the bug. Every message published before that time is marked acknowledged, and every message after it unacknowledged, so those are delivered again.
Let's build the bad window. orders-loader processes four orders. Then release v2 is deployed, mark.py saves that time, and the buggy release processes three more. We run python publish_orders.py 101 104, python drain.py orders-loader, python mark.py, python publish_orders.py 105 107 and python drain.py orders-loader.
$ python publish_orders.py 101 104
published order-101 to order-104
$ python drain.py orders-loader
orders-loader got 4 | order-101, order-102, order-103, order-104
$ python mark.py
release v2 deployed, time saved to release_time.txt
$ python publish_orders.py 105 107
published order-105 to order-107
$ python drain.py orders-loader
orders-loader got 3 | order-105, order-106, order-107The first drain gets order-101 to order-104, the second gets order-105, order-106, order-107. Now the fix is in place, so we seek to the release time with python seek.py orders-loader --time and drain again.
$ python seek.py orders-loader --time
seek orders-loader to the release time
$ python drain.py orders-loader
orders-loader got 3 | order-105, order-106, order-107Exactly the three orders from the bad window come back: orders-loader got 3 | order-105, order-106, order-107.
Replaying acked messages only works if they are still stored: topic retention, as on this topic, or retain acked messages on the subscription. And seeking to a time in the future does the opposite: it purges the backlog.
Seek a little earlier than the bug, because publish times are server times and not exact, and a seek can take about a minute to take full effect. So the consumer must handle duplicates.
Question 3: Someone asks how long Pub/Sub keeps messages. How do message retention and topic retention work?
There are two settings. Subscription retention keeps unacked messages: 7 days by default, from 10 minutes up to 31 days. Topic retention is optional, also 10 minutes to 31 days, and it keeps every message, acked or not.
| Setting | Default | Range | Keeps |
|---|---|---|---|
| Subscription retention | 7 days | 10 minutes to 31 days | unacked messages |
| Topic retention | off | 10 minutes to 31 days | every message, acked or not |
| Retain acked messages | off | uses the subscription retention | acked messages too, on that subscription |
Let's read the settings from our setup and try two topic retention values with python retention.py.
$ python retention.py
topic orders, retention: 7 days, 0:00:00
subscription orders-loader, retention: 7 days, 0:00:00
retain acked messages: False
31 days of topic retention: accepted
32 days of topic retention: InvalidArgument 400 message_retention_duration out of boundsThe orders topic has 7 days, 0:00:00, orders-loader shows the 7 day default, and retain acked messages: False. 31 days is accepted, and 32 days is rejected: message_retention_duration out of bounds.
One honest note: the emulator checks these values, but doesn't apply them. It keeps every message, so we can't show expiry here. On Google Cloud, messages older than the window are gone.
Cost matters too. Keeping unacked messages longer than a day adds storage charges, unless the topic already retains them. Topic retention is the cheapest replay option, because all subscriptions share it.
By default, a subscription with no activity for 31 days is deleted. For a pipeline that runs rarely, set it to never expire.
gcloud pubsub topics create orders \
--message-retention-duration=7d
gcloud pubsub subscriptions create orders-loader \
--topic=orders \
--message-retention-duration=7d \
--expiration-period=neverQuestion 4: A teammate wants to reprocess from a known point before a deploy. How do snapshots help?
A snapshot saves a subscription's acknowledgement state at one point in time. Later, we seek the subscription to it, and every message that was unacked then, or published after it, is delivered again.
Before the deploy, let's snapshot orders-loader as before-deploy with python snapshot.py orders-loader before-deploy.
$ python snapshot.py orders-loader before-deploy
snapshot before-deploy of topic orders
expires in 7 daysIt belongs to the topic orders, and it expires in 7 days.
Now the new release processes orders 108 to 110, and it's wrong. We run python publish_orders.py 108 110 and python drain.py orders-loader, then seek back with python seek.py orders-loader before-deploy and drain again.
$ python publish_orders.py 108 110
published order-108 to order-110
$ python drain.py orders-loader
orders-loader got 3 | order-108, order-109, order-110
$ python seek.py orders-loader before-deploy
seek orders-loader to snapshot before-deploy
$ python drain.py orders-loader
orders-loader got 3 | order-108, order-109, order-110The first drain gets order-108, order-109, order-110. After seek orders-loader to snapshot before-deploy, the same three orders are delivered again.
The limits: a snapshot lives at most 7 days, minus the age of the oldest unacked message, and Pub/Sub refuses one that would expire in under 1 hour. A project can hold 5,000. If there is a backlog, creating a snapshot has a one-time fee, equal to 7 days of storing that backlog.
The release pipeline takes a snapshot before every consumer deploy.
Question 5: Acked messages are needed again for a new consumer. What are the options?
By default, acked messages are discarded, so this only works if we planned for it. There are three options:
- A snapshot taken earlier - any subscription on the topic can seek to it, even a new one
retain_acked_messages- on one subscription: seek it back in time- Topic retention - every message, acked or not, shared by all subscriptions
For a brand-new consumer, a snapshot is the cleanest. Let's create the fraud-model subscription, seek it to before-deploy, and drain it. We run python new_sub.py fraud-model orders before-deploy and python drain.py fraud-model.
$ python new_sub.py fraud-model orders before-deploy
created fraud-model filter: '' | seek to before-deploy
$ python drain.py fraud-model
fraud-model got 3 | order-108, order-109, order-110It prints fraud-model got 3 | order-108, order-109, order-110, although orders-loader acked them already.
Retain acked messages keeps acked messages on one subscription for its retention window. We can seek that subscription back in time, but only its own consumers see the replay. Topic retention keeps every message on the topic for its window, for all subscriptions. It's the cheapest replay option, and the one I turn on for every important topic.
The acked data is gone from Pub/Sub. Reload it from the source system or the data lake.
Question 6: A filter was added and old messages are still delivered. What are the rules for changing filters?
The rule: we set a filter when we create a subscription, and it cannot be changed or removed after that. So the old subscription keeps delivering everything.
Let's try anyway. payments-all gets a backlog of six new events, and we try to give it a refund filter. We run python publish_payments.py 107 and python change_filter.py.
$ python publish_payments.py 107
published 6 events: payment, refund, payment, chargeback, refund, payment
$ python change_filter.py
InvalidArgument 400 Updating the filter field is currently unsupported in the Pub/Sub Emulator.The update is rejected: Updating the filter field is currently unsupported in the Pub/Sub Emulator. Google Cloud rejects it too, because the filter is immutable. A new filter means a new subscription. The safe way:
- First, snapshot the old subscription
- Then create the new subscription with the filter
- Seek it to the snapshot
- Move the consumers, then delete the old subscription
Let's do the first three steps for refunds-audit. We run python snapshot.py payments-all move-to-refunds, then python new_sub.py with the refund filter, and drain the new subscription.
$ python snapshot.py payments-all move-to-refunds
snapshot move-to-refunds of topic payments
expires in 7 days
$ python new_sub.py refunds-audit payments move-to-refunds 'attributes.event_type = "refund"'
created refunds-audit filter: 'attributes.event_type = "refund"' | seek to move-to-refunds
$ python drain.py refunds-audit
refunds-audit got 2 | txn-108 refund, txn-111 refundFrom that backlog of six, refunds-audit got 2 | txn-108 refund, txn-111 refund. Seeking redelivers only messages that match the new subscription's filter. While both subscriptions exist, refunds arrive on both, so the consumers must be idempotent.
Review the filter like code before you create the subscription, because a typo means doing this migration again.
Scenario 7: A bad release corrupted a downstream table for three hours. How do you recover the data with Pub/Sub?
First, stop the damage: roll back or fix the consumer. Then check that topic retention or retain acked messages still holds those three hours. Our sales topic has 7 days of topic retention, and loader.py upserts into a sales table by order_id.
Release v1 loads three sales. We run python publish_sales.py 201 203 and python loader.py v1.
$ python publish_sales.py 201 203
published sales for orders 201 to 203
$ python loader.py v1
v1 loaded 3 events
(201, 20.05)
(202, 20.1)
(203, 20.15)The rows are correct: (201, 20.05), (202, 20.1), (203, 20.15). Now release v2 is deployed at a known time, and it loads the next three. We run python mark.py, python publish_sales.py 204 206 and python loader.py v2.
$ python mark.py
release v2 deployed, time saved to release_time.txt
$ python publish_sales.py 204 206
published sales for orders 204 to 206
$ python loader.py v2
v2 loaded 3 events
(201, 20.05)
(202, 20.1)
(203, 20.15)
(204, 2020.0)
(205, 2025.0)
(206, 2030.0)Release v2 stores the amounts in cents, not dollars: (204, 2020.0) instead of 20.2. With the fix deployed, we seek to the release time with python seek.py sales-to-table --time and run the fixed loader, python loader.py v3.
$ python seek.py sales-to-table --time
seek sales-to-table to the release time
$ python loader.py v3
v3 loaded 3 events
(201, 20.05)
(202, 20.1)
(203, 20.15)
(204, 20.2)
(205, 20.25)
(206, 20.3)v3 loaded 3 events, and the upsert overwrites the bad rows: still six rows, all correct, with (204, 20.2), (205, 20.25) and (206, 20.3).
A replay means duplicates, so this only works because the consumer is idempotent:
- Idempotent consumer - upsert or MERGE on a business key, never a blind insert
- Dataflow - don't seek a running pipeline: seek a new subscription, drain or cancel, resubmit on it
- Next time - snapshot before every release, retention longer than detection time, data checks
Take a snapshot before every release, keep topic retention longer than it takes you to notice, and add a data check on the table.
Key points to remember
- Filters: attributes only, set at creation, can't be changed
- Seek to a time: before = acked, after = delivered again
- Subscription retention 7 days default, 10 minutes to 31 days; topic retention optional
- Snapshots: up to 7 days, any subscription on the topic can seek to one
- Replay needs retention + an idempotent consumer
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 #7 - Filtering, replay and retention
# 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_07_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
from google.protobuf.duration_pb2 import Duration
project = "tb-shop-dev"
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
week = Duration(seconds=7 * 24 * 3600)
for topic, name in [("payments", "payments-all"),
("orders", "orders-loader"),
("sales", "sales-to-table")]:
path = pub.topic_path(project, topic)
pub.create_topic(request={"name": path,
"message_retention_duration": week})
sub.create_subscription(
name=sub.subscription_path(project, name), topic=path)
""")
write('refunds.py', """\
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
s = sub.create_subscription(request={
"name": sub.subscription_path("tb-shop-dev", "refunds-only"),
"topic": "projects/tb-shop-dev/topics/payments",
"filter": 'attributes.event_type = "refund"',
})
print("created", s.name.split("/")[-1], "with filter:", s.filter)
""")
write('publish_payments.py', """\
import sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "payments")
kinds = ["payment", "refund", "payment",
"chargeback", "refund", "payment"]
for n, kind in enumerate(kinds, int(sys.argv[1])):
pub.publish(topic, f"txn-{n}".encode(), event_type=kind).result()
print("published", len(kinds), "events:", ", ".join(kinds))
""")
write('drain.py', """\
import sys
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", sys.argv[1])
got = []
while True: # pull until nothing is left, ack every batch
resp = sub.pull(subscription=path, max_messages=100, timeout=10)
if not resp.received_messages:
break
for m in resp.received_messages:
text = m.message.data.decode()
kind = m.message.attributes.get("event_type", "")
got.append(f"{text} {kind}".strip())
ids = [m.ack_id for m in resp.received_messages]
sub.acknowledge(subscription=path, ack_ids=ids)
print(sys.argv[1], "got", len(got), "|", ", ".join(sorted(got)))
""")
write('publish_orders.py', """\
import sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "orders")
first, last = int(sys.argv[1]), int(sys.argv[2])
for n in range(first, last + 1):
pub.publish(topic, f"order-{n}".encode()).result()
print(f"published order-{first} to order-{last}")
""")
write('mark.py', """\
import datetime, time
time.sleep(1.5) # a clear gap before and after the mark
now = datetime.datetime.now(datetime.timezone.utc)
open("release_time.txt", "w").write(now.isoformat())
time.sleep(1.5)
print("release v2 deployed, time saved to release_time.txt")
""")
write('seek.py', """\
import datetime, sys
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", sys.argv[1])
if sys.argv[2] == "--time":
when = open("release_time.txt").read()
sub.seek(request={"subscription": path,
"time": datetime.datetime.fromisoformat(when)})
print("seek", sys.argv[1], "to the release time")
else:
snap = sub.snapshot_path("tb-shop-dev", sys.argv[2])
sub.seek(request={"subscription": path, "snapshot": snap})
print("seek", sys.argv[1], "to snapshot", sys.argv[2])
""")
write('retention.py', """\
from google.cloud import pubsub_v1
from google.protobuf.duration_pb2 import Duration
pub = pubsub_v1.PublisherClient()
sub = pubsub_v1.SubscriberClient()
t = pub.get_topic(topic=pub.topic_path("tb-shop-dev", "orders"))
print("topic orders, retention:", t.message_retention_duration)
s = sub.get_subscription(subscription=sub.subscription_path(
"tb-shop-dev", "orders-loader"))
print("subscription orders-loader, retention:",
s.message_retention_duration)
print("retain acked messages:", s.retain_acked_messages)
for days in (31, 32):
try:
keep = Duration(seconds=days * 86400)
pub.create_topic(request={
"name": pub.topic_path("tb-shop-dev", f"clicks-{days}d"),
"message_retention_duration": keep})
print(days, "days of topic retention: accepted")
except Exception as e:
print(days, "days of topic retention:", type(e).__name__, e)
""")
write('snapshot.py', """\
import datetime, sys
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", sys.argv[1])
snap = sub.create_snapshot(request={
"name": sub.snapshot_path("tb-shop-dev", sys.argv[2]),
"subscription": path})
now = datetime.datetime.now(datetime.timezone.utc)
left = snap.expire_time - now
print("snapshot", sys.argv[2], "of topic", snap.topic.split("/")[-1])
print("expires in", round(left.total_seconds() / 86400), "days")
""")
write('new_sub.py', """\
import sys
from google.cloud import pubsub_v1
project = "tb-shop-dev"
name, topic, snapshot = sys.argv[1:4]
flt = sys.argv[4] if len(sys.argv) > 4 else ""
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path(project, name)
sub.create_subscription(request={
"name": path, "topic": f"projects/{project}/topics/{topic}",
"filter": flt})
sub.seek(request={"subscription": path,
"snapshot": sub.snapshot_path(project, snapshot)})
print("created", name, "filter:", repr(flt), "| seek to", snapshot)
""")
write('change_filter.py', """\
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "payments-all")
refunds = 'attributes.event_type = "refund"'
try:
sub.update_subscription(request={
"subscription": {"name": path, "filter": refunds},
"update_mask": {"paths": ["filter"]}}, timeout=10, retry=None)
print("filter changed")
except Exception as e:
print(type(e).__name__, e)
""")
write('publish_sales.py', """\
import json, sys
from google.cloud import pubsub_v1
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", "sales")
first, last = int(sys.argv[1]), int(sys.argv[2])
for n in range(first, last + 1):
event = {"order_id": n, "amount_cents": 1000 + n * 5}
pub.publish(topic, json.dumps(event).encode()).result()
print("published sales for orders", first, "to", last)
""")
write('loader.py', """\
import json, sqlite3, sys
from google.cloud import pubsub_v1
release = sys.argv[1]
db = sqlite3.connect("warehouse.db")
db.execute("CREATE TABLE IF NOT EXISTS sales"
" (order_id INTEGER PRIMARY KEY, amount REAL)")
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "sales-to-table")
UPSERT = ("INSERT INTO sales VALUES (?, ?) ON CONFLICT(order_id)"
" DO UPDATE SET amount = excluded.amount")
n = 0
while True:
resp = sub.pull(subscription=path, max_messages=100, timeout=10)
if not resp.received_messages:
break
for m in resp.received_messages:
e = json.loads(m.message.data)
cents = e["amount_cents"]
amount = cents if release == "v2" else cents / 100 # v2 bug
db.execute(UPSERT, (e["order_id"], amount))
n += 1
ids = [m.ack_id for m in resp.received_messages]
db.commit() # write first, then ack
sub.acknowledge(subscription=path, ack_ids=ids)
print(release, "loaded", n, "events")
for row in db.execute("SELECT * FROM sales ORDER BY order_id"):
print(row)
""")
run('python setup.py')
# Step 1: A filtered subscription next to an unfiltered one
run('python refunds.py')
run('python publish_payments.py 101')
run('python drain.py payments-all')
run('python drain.py refunds-only')
# Step 2: Four good orders, then release v2 processes three more
run('python publish_orders.py 101 104')
run('python drain.py orders-loader')
run('python mark.py')
run('python publish_orders.py 105 107')
run('python drain.py orders-loader')
# Step 3: Seek back to the release time
run('python seek.py orders-loader --time')
run('python drain.py orders-loader')
# Step 4: Retention settings (accepted by the emulator, not applied)
run('python retention.py')
# Step 5: Snapshot before the deploy
run('python snapshot.py orders-loader before-deploy')
# Step 6: The new release processes three orders, then we seek back to the snapshot
run('python publish_orders.py 108 110')
run('python drain.py orders-loader')
run('python seek.py orders-loader before-deploy')
run('python drain.py orders-loader')
# Step 7: A brand-new subscription replays from the snapshot
run('python new_sub.py fraud-model orders before-deploy')
run('python drain.py fraud-model')
# Step 8: A backlog on payments-all, then try to add a filter to it
run('python publish_payments.py 107')
run('python change_filter.py')
# Step 9: Snapshot, new filtered subscription, seek
run('python snapshot.py payments-all move-to-refunds')
run('python new_sub.py refunds-audit payments move-to-refunds \'attributes.event_type = "refund"\'')
run('python drain.py refunds-audit')
# Step 10: Release v1 loads three sales
run('python publish_sales.py 201 203')
run('python loader.py v1')
# Step 11: Release v2 is deployed and loads three more
run('python mark.py')
run('python publish_sales.py 204 206')
run('python loader.py v2')
# Step 12: Fix deployed: seek to the release time and reload
run('python seek.py sales-to-table --time')
run('python loader.py v3')
Part 8: Schemas and Data Contracts
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.