In this post we are going to cover Pub/Sub subscriptions: pull and push. 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 Cloud Run service and a batch job both need the same events with very different speeds, and you design the subscriptions.
A subscription is how a consumer gets messages from a topic. With pull, our workers ask Pub/Sub for messages. With push, Pub/Sub calls our HTTPS endpoint. Interviewers ask about this because the choice drives everything after it: flow control, ack deadlines, retries, scaling and monitoring. If you can explain why a message came back, or why a push endpoint got flooded, you show that you have run consumers in production.
Part 3 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 new service must consume order events. How do you choose between pull and push subscriptions?
- Question 2: The pull consumer is slow and keeps falling behind. How does streaming pull with flow control help?
- Question 3: A teammate asks what the ack deadline is and why messages come back. How do you explain it?
- Question 4: A push endpoint returns errors and gets flooded with retries. How does push backoff work?
- Question 5: Several workers read from one subscription. How does Pub/Sub share messages between them?
- Question 6: Your manager asks how to see how far behind a subscriber is. Which metrics do you check?
- Scenario 7: A Cloud Run service and a batch job both need the same events with very different speeds. How do you design the subscriptions?
- The full script
The short answer
A pull subscription lets your workers ask for messages and control flow and deadlines; a push subscription posts each message to an HTTPS endpoint, and the response code is the ack. Streaming pull with flow control keeps a consumer from drowning, the ack deadline decides when a message comes back, and oldest unacked message age tells you how far behind a subscriber is.
Let's set up the project
We run Google's official Pub/Sub emulator locally, with small Python scripts that use the google-cloud-pubsub client, so every output is real. Features the emulator doesn't support are shown as code. Before the first step we run python setup.py once. It creates the orders, clicks, invoices, refunds and payments topics in the project tb-shop-dev, with the pull subscriptions analytics-pull, invoices-pull, ledger-pull and audit-pull. endpoint.py is a tiny local HTTP server that stands in for a push endpoint like Cloud Run.
from google.cloud import pubsub_v1
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topics = {"orders": [], "clicks": ["analytics-pull"],
"invoices": ["invoices-pull"], "refunds": [],
"payments": ["ledger-pull", "audit-pull"]}
for topic_id, subs in topics.items():
topic = pub.topic_path("tb-shop-dev", topic_id)
pub.create_topic(name=topic)
for sub_id in subs:
path = sub.subscription_path("tb-shop-dev", sub_id)
sub.create_subscription(name=path, topic=topic)import json, sys
from google.cloud import pubsub_v1
topic_id = sys.argv[1]
count, start = int(sys.argv[2]), int(sys.argv[3])
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", topic_id)
for order_id in range(start, start + count):
data = json.dumps({"order_id": order_id}).encode()
message_id = pub.publish(topic, data).result() # wait for each
if count == 1:
print("published order", order_id, "as message", message_id)
if count > 1:
print("published", count, "orders to", topic_id)import json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
def start(codes=(204,)):
"""A tiny push endpoint on a free local port, like a Cloud Run
service. Request N gets codes[N]; the last code repeats."""
got = []
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
size = int(self.headers["Content-Length"])
body = json.loads(self.rfile.read(size))
code = codes[min(len(got), len(codes) - 1)]
got.append((body, code))
self.send_response(code)
self.end_headers()
def log_message(self, *args): # keep the terminal quiet
pass
server = ThreadingHTTPServer(("localhost", 0), Handler) # free port
t = threading.Thread(target=server.serve_forever, daemon=True)
t.start()
url = f"http://localhost:{server.server_port}/push"
return server, url, gotimport sys
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", sys.argv[1])
total = 0
while True:
resp = sub.pull(subscription=path, max_messages=100, timeout=10)
if not resp.received_messages:
break
total += len(resp.received_messages)
ack_ids = [r.ack_id for r in resp.received_messages]
sub.acknowledge(subscription=path, ack_ids=ack_ids)
print(sys.argv[1], "pulled and acked", total, "messages")Question 1: A new service must consume order events. How do you choose between pull and push subscriptions?
Start from who opens the connection. With pull, our workers call Pub/Sub, so they control flow, deadlines and scaling, and need no public endpoint. With push, Pub/Sub posts each message to our HTTPS endpoint, like a Cloud Run service or a webhook.
Let's try both on one topic. In pull_and_push.py we create two subscriptions on orders: shipping-pull, and billing-push. The only difference is the push config with an endpoint URL. Then we publish one order and run python pull_and_push.py.
import base64, time
from google.cloud import pubsub_v1
import endpoint
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "orders")
pull_sub = sub.subscription_path("tb-shop-dev", "shipping-pull")
push_sub = sub.subscription_path("tb-shop-dev", "billing-push")
server, url, got = endpoint.start() # stands in for Cloud Run
sub.create_subscription(name=pull_sub, topic=topic)
sub.create_subscription(name=push_sub, topic=topic,
push_config={"push_endpoint": url})
pub.publish(topic, b'{"order_id": 1001}').result()
resp = sub.pull(subscription=pull_sub, max_messages=10) # we ask
for r in resp.received_messages:
print("pull:", r.message.message_id, r.message.data.decode())
sub.acknowledge(subscription=pull_sub, ack_ids=[r.ack_id])
time.sleep(3) # Pub/Sub calls us
for body, code in got:
data = base64.b64decode(body["message"]["data"]).decode()
print("push:", body["message"]["messageId"], data,
"from", body["subscription"].split("/")[-1], "->", code)
sub.delete_subscription(subscription=push_sub)$ python pull_and_push.py
pull: 1 {"order_id": 1001}
push: 1 {"order_id": 1001} from billing-push -> 204The pull subscription returns pull: 1 {"order_id": 1001} when we ask, and we ack it. The push subscription posts the same message to the endpoint: push: 1 {"order_id": 1001} from billing-push -> 204. The message arrives wrapped in JSON, with the data in base64.
For push, the response is the ack:
- push ack = 102, 200, 201, 202, 204
- any other response = nack
In production, the endpoint must be public HTTPS with a CA-signed certificate, and no private VPC addresses. The emulator accepts plain http://localhost.
Streaming pull for high-volume pipelines, tight flow control, or exactly-once delivery, which works only with pull. Push when the consumer is serverless or can't run the client library.
Question 2: The pull consumer is slow and keeps falling behind. How does streaming pull with flow control help?
Streaming pull keeps one long-lived connection open, and messages stream in as they arrive. Flow control is the brake: when outstanding messages or bytes hit the limit, the client stops taking more until we ack or nack.
Let's try it. We publish 20 orders to clicks, and stream.py has a callback that takes 200 milliseconds. First we run it with python stream.py 3, so max messages is 3. Then we publish 20 more and run python stream.py 1000, the Python client's default.
import sys, threading, time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import FlowControl
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "analytics-pull")
lock, s = threading.Lock(), {"now": 0, "peak": 0, "done": 0}
def callback(message):
with lock:
s["now"] += 1
s["peak"] = max(s["peak"], s["now"])
time.sleep(0.2) # slow work: 200 ms
message.ack()
with lock:
s["now"] -= 1
s["done"] += 1
flow = FlowControl(max_messages=int(sys.argv[1])) # the brake
future = sub.subscribe(path, callback=callback, flow_control=flow)
try:
future.result(timeout=6) # stream for 6 seconds
except TimeoutError:
future.cancel()
future.result()
print(f"max_messages={flow.max_messages}: processed {s['done']},",
f"peak in flight {s['peak']}")$ python publish.py clicks 20 2001
published 20 orders to clicks
$ python stream.py 3
max_messages=3: processed 20, peak in flight 3
$ python publish.py clicks 20 3001
published 20 orders to clicks
$ python stream.py 1000
max_messages=1000: processed 20, peak in flight 10With the brake on: max_messages=3: processed 20, peak in flight 3. The consumer works at its own pace instead of drowning.
With the default: max_messages=1000: processed 20, peak in flight 10. The peak is 10 because of the default pool of 10 callback threads in the Python client. The rest of the leased messages wait in memory, while the client keeps extending their deadlines. The default limits are 1,000 messages and 100 MiB.
So find the real bottleneck. For slow I/O, a bigger thread pool helps. For CPU or memory, lower max messages and add workers: one subscription scales out across many clients.
One stream carries up to 10 MB per second. At very high volume, run more subscriber clients, not one giant one.
Question 3: A teammate asks what the ack deadline is and why messages come back. How do you explain it?
The ack deadline is how long Pub/Sub waits for an ack after a delivery. No ack in time, and it delivers the message again. The default is 10 seconds, and we can set 10 to 600.
Let's publish order 5001 to invoices and run python deadline.py. It pulls the message without an ack, pulls again before the deadline, then waits past the deadline and pulls once more.
import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "invoices-pull")
config = sub.get_subscription(subscription=path)
print("ack deadline:", config.ack_deadline_seconds, "seconds")
resp = sub.pull(subscription=path, max_messages=1)
first = resp.received_messages[0]
print("pulled message", first.message.message_id, "- no ack")
start = time.time()
time.sleep(2)
early = sub.pull(subscription=path, max_messages=1, timeout=6)
n = len(early.received_messages)
print("pull before the deadline:", n, "messages")
time.sleep(max(0, 13 - (time.time() - start)))
resp = sub.pull(subscription=path, max_messages=1)
again = resp.received_messages[0]
print("pull after the deadline: message", again.message.message_id)
same = again.message.message_id == first.message.message_id
print("same message id:", same)
print("same ack id:", again.ack_id == first.ack_id)
sub.acknowledge(subscription=path, ack_ids=[again.ack_id])$ python publish.py invoices 1 5001
published order 5001 as message 42
$ python deadline.py
ack deadline: 10 seconds
pulled message 42 - no ack
pull before the deadline: 0 messages
pull after the deadline: message 42
same message id: True
same ack id: FalseThe subscription has ack deadline: 10 seconds. We get pulled message 42 - no ack. A pull before the deadline gets 0 messages, because it's leased to us. After the deadline it's back: pull after the deadline: message 42, with same message id: True and same ack id: False. So always ack with the ack ID from the latest delivery.
A message comes back for one of three reasons:
- Deadline expired - slow work, or a dead worker
- Ack lost - on the way back
- Nack - the subscriber gives it back
Without exactly-once delivery, deadlines are best effort, so plan for duplicates.
Raising the deadline to 600 to fix slow work. The client library already extends deadlines while it holds a message, up to one hour by default (max_lease_duration). The Python streaming client sets its own lease deadlines anyway, and with unary pull a 600 second deadline mostly delays redelivery when a worker dies.
Question 4: A push endpoint returns errors and gets flooded with retries. How does push backoff work?
Push treats every response that isn't a success code as a nack. To protect the endpoint, Pub/Sub applies push backoff: exponential, from 100 milliseconds up to 60 seconds, for the whole subscription, and you can't turn it off.
Let's make an endpoint that fails. In push_retry.py it answers 500, 503, 429 and 400, then 204. We publish one refund and run python push_retry.py.
import time
from google.cloud import pubsub_v1
import endpoint
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "refunds")
path = sub.subscription_path("tb-shop-dev", "refunds-push")
server, url, got = endpoint.start(codes=[500, 503, 429, 400, 204])
sub.create_subscription(name=path, topic=topic,
push_config={"push_endpoint": url})
pub.publish(topic, b'{"refund_id": 77}').result()
time.sleep(10)
for n, (body, code) in enumerate(got, 1):
message_id = body["message"]["messageId"]
print("attempt", n, "message", message_id, "->", code)
sub.delete_subscription(subscription=path)$ python push_retry.py
attempt 1 message 43 -> 500
attempt 2 message 43 -> 503
attempt 3 message 43 -> 429
attempt 4 message 43 -> 400
attempt 5 message 43 -> 204The same message, message 43, takes five attempts: -> 500, -> 503, -> 429, -> 400, and then attempt 5 message 43 -> 204. Even the 400 is retried, because push can't tell bad data from an outage.
One honest note: the emulator retries about every second, with no backoff. The real service backs off. A retry policy on the subscription adds its own backoff, and the larger of the two delays wins.
Pub/Sub also uses a push window:
- Push window - starts small, grows while the endpoint acks quickly
- Shrinks - on failures, or latency over 1 second
So a struggling endpoint gets less traffic, not more.
Return success only after the work is done, and add a dead-letter topic, so one bad message can't fail forever. And for push, the ack deadline is also the request timeout.
Question 5: Several workers read from one subscription. How does Pub/Sub share messages between them?
Workers on the same subscription compete: Pub/Sub hands each message to one of them, so we scale out just by starting more clients. That's different from fan-out, where every subscription gets its own copy.
Let's publish 30 orders to payments. Then python workers.py starts three subscriber clients, like three pods, streaming from ledger-pull, each with max messages 2. Last, python drain.py audit-pull reads the second subscription.
import collections, threading, time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import FlowControl
path = "projects/tb-shop-dev/subscriptions/ledger-pull"
lock, seen = threading.Lock(), collections.Counter()
def callback(message):
time.sleep(0.1) # the work
with lock:
seen[message.data.decode()] += 1
message.ack()
workers = []
for _ in range(3): # three subscriber clients
client = pubsub_v1.SubscriberClient()
workers.append(client.subscribe(path, callback=callback,
flow_control=FlowControl(max_messages=2)))
time.sleep(8)
for w in workers:
w.cancel()
print("3 workers on ledger-pull: processed", sum(seen.values()),
"messages,", len(seen), "unique orders")
print("most deliveries of one order:", max(seen.values()))$ python publish.py payments 30 6001
published 30 orders to payments
$ python workers.py
3 workers on ledger-pull: processed 30 messages, 30 unique orders
most deliveries of one order: 1
$ python drain.py audit-pull
audit-pull pulled and acked 30 messagesWe get processed 30 messages, 30 unique orders, and most deliveries of one order: 1, so no order was delivered twice in this run. The split between workers depends on load and timing, so we don't print it. And audit-pull pulled and acked 30 messages: each subscription is an independent copy of the stream.
Duplicates can still happen, for example when a slow worker's deadline expires, so consumers must be idempotent. That's the next part of this series.
With ordering keys, only one batch per key is outstanding at a time, so one hot key can't spread across workers. And with Dataflow, give every pipeline its own subscription; never share one.
Question 6: Your manager asks how to see how far behind a subscriber is. Which metrics do you check?
Two metrics. Oldest unacked message age says how far behind we are in time, and the number of unacked messages says how big the backlog is. The docs recommend the by-region versions of both.
Age is the one I alert on. A big backlog that drains quickly is fine, but an old message means a consumer is stuck, or one message keeps failing.
Then find out why. Here is the list we check.
| Metric (pubsub.googleapis.com/...) | Tells you |
|---|---|
| subscription/oldest_unacked_message_age_by_region | how far behind, in time |
| subscription/num_unacked_messages_by_region | how big the backlog is |
| subscription/expired_ack_deadlines_count | subscribers too slow for the deadline |
| subscription/push_request_count (response_code) | push endpoint errors |
| subscription/dead_letter_message_count | messages that gave up |
| topic/send_request_count | how fast publishers send |
Expired ack deadlines count means subscribers are too slow for their deadline. Push request count, by response code, shows endpoint errors. Dead letter message count shows messages that gave up.
Then compare with the topic's send request count. If publishing went up, add subscribers. If it's flat, the consumer got slower.
The emulator has no metrics, so this part is Cloud Monitoring only. Alert on oldest unacked message age per subscription, with a threshold from your latency target.
Scenario 7: A Cloud Run service and a batch job both need the same events with very different speeds. How do you design the subscriptions?
One topic, two subscriptions. Each subscription keeps its own backlog and its own acks, so the slow consumer never holds back the fast one.
Let's build it with python scenario.py. Cloud Run gets a push subscription, shipping-run, with a 60 second ack deadline. The batch job gets a pull subscription, finance-batch. Then we publish five orders, 7001 to 7005.
import time
from google.cloud import pubsub_v1
import endpoint
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "orders")
fast = sub.subscription_path("tb-shop-dev", "shipping-run")
slow = sub.subscription_path("tb-shop-dev", "finance-batch")
server, url, got = endpoint.start() # stands in for Cloud Run
sub.create_subscription(name=fast, topic=topic,
push_config={"push_endpoint": url},
ack_deadline_seconds=60)
sub.create_subscription(name=slow, topic=topic)
for order_id in range(7001, 7006):
data = f'{{"order_id": {order_id}}}'.encode()
pub.publish(topic, data).result()
time.sleep(3)
print("shipping-run (push):", len(got), "orders reached the service")
print("finance-batch (pull): waiting for the batch job")
sub.delete_subscription(subscription=fast)$ python scenario.py
shipping-run (push): 5 orders reached the service
finance-batch (pull): waiting for the batch jobshipping-run (push): 5 orders reached the service within seconds, and finance-batch (pull): waiting for the batch job. The same five orders wait in its backlog. Later, the batch job runs with python drain.py finance-batch.
$ python drain.py finance-batch
finance-batch pulled and acked 5 messagesfinance-batch pulled and acked 5 messages.
In production, push to the Cloud Run URL with an OIDC token. Its service account needs the run invoker role, and the Pub/Sub service agent needs token creator on it. Set the ack deadline to cover the request: for push, it's also the timeout.
For the batch subscription, retention must cover the longest gap between runs: 7 days by default, up to 31. And set expiration to never, because by default a subscription with no activity for 31 days is deleted.
SA=push-sa@tb-shop-dev.iam.gserviceaccount.com
PSA=service-123456789@gcp-sa-pubsub.iam.gserviceaccount.com
RUN_URL=https://shipping-123456789.us-central1.run.app
gcloud run services add-iam-policy-binding shipping \
--member=serviceAccount:$SA --role=roles/run.invoker \
--region=us-central1
gcloud iam service-accounts add-iam-policy-binding $SA \
--member=serviceAccount:$PSA \
--role=roles/iam.serviceAccountTokenCreator
gcloud pubsub subscriptions create shipping-run --topic=orders \
--push-endpoint=$RUN_URL/orders --ack-deadline=60 \
--push-auth-service-account=$SA --push-auth-token-audience=$RUN_URL
gcloud pubsub subscriptions create finance-batch --topic=orders \
--message-retention-duration=7d --expiration-period=neverUnacked messages kept longer than a day add storage charges. And alert when the batch backlog's oldest message is older than one run interval, so a skipped run gets noticed.
Key points to remember
- Pull for volume and control; push for Cloud Run and webhooks
- Streaming pull + FlowControl: the client takes only what it can handle
- No ack before the ack deadline (10 to 600 s): the message comes back
- Push treats errors as nacks, with exponential backoff on the real service
- Workers share a subscription; watch oldest unacked message age
The full script
This is a Python script that writes every file and runs every command above in order against the Pub/Sub emulator: run pip install google-cloud-pubsub first.
Show the full script that writes the files and runs every step
# ========================================================================
# Pub/Sub Interview Q&A #3 - Subscriptions: pull and push
# 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_03_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
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topics = {"orders": [], "clicks": ["analytics-pull"],
"invoices": ["invoices-pull"], "refunds": [],
"payments": ["ledger-pull", "audit-pull"]}
for topic_id, subs in topics.items():
topic = pub.topic_path("tb-shop-dev", topic_id)
pub.create_topic(name=topic)
for sub_id in subs:
path = sub.subscription_path("tb-shop-dev", sub_id)
sub.create_subscription(name=path, topic=topic)
""")
write('publish.py', """\
import json, sys
from google.cloud import pubsub_v1
topic_id = sys.argv[1]
count, start = int(sys.argv[2]), int(sys.argv[3])
pub = pubsub_v1.PublisherClient()
topic = pub.topic_path("tb-shop-dev", topic_id)
for order_id in range(start, start + count):
data = json.dumps({"order_id": order_id}).encode()
message_id = pub.publish(topic, data).result() # wait for each
if count == 1:
print("published order", order_id, "as message", message_id)
if count > 1:
print("published", count, "orders to", topic_id)
""")
write('endpoint.py', 'import json, threading\nfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServer\n\n\ndef start(codes=(204,)):\n """A tiny push endpoint on a free local port, like a Cloud Run\n service. Request N gets codes[N]; the last code repeats."""\n got = []\n\n class Handler(BaseHTTPRequestHandler):\n def do_POST(self):\n size = int(self.headers["Content-Length"])\n body = json.loads(self.rfile.read(size))\n code = codes[min(len(got), len(codes) - 1)]\n got.append((body, code))\n self.send_response(code)\n self.end_headers()\n\n def log_message(self, *args): # keep the terminal quiet\n pass\n\n server = ThreadingHTTPServer(("localhost", 0), Handler) # free port\n t = threading.Thread(target=server.serve_forever, daemon=True)\n t.start()\n url = f"http://localhost:{server.server_port}/push"\n return server, url, got\n')
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])
total = 0
while True:
resp = sub.pull(subscription=path, max_messages=100, timeout=10)
if not resp.received_messages:
break
total += len(resp.received_messages)
ack_ids = [r.ack_id for r in resp.received_messages]
sub.acknowledge(subscription=path, ack_ids=ack_ids)
print(sys.argv[1], "pulled and acked", total, "messages")
""")
run('python setup.py')
# Step 1: One topic, a pull and a push subscription, one order
write('pull_and_push.py', """\
import base64, time
from google.cloud import pubsub_v1
import endpoint
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "orders")
pull_sub = sub.subscription_path("tb-shop-dev", "shipping-pull")
push_sub = sub.subscription_path("tb-shop-dev", "billing-push")
server, url, got = endpoint.start() # stands in for Cloud Run
sub.create_subscription(name=pull_sub, topic=topic)
sub.create_subscription(name=push_sub, topic=topic,
push_config={"push_endpoint": url})
pub.publish(topic, b'{"order_id": 1001}').result()
resp = sub.pull(subscription=pull_sub, max_messages=10) # we ask
for r in resp.received_messages:
print("pull:", r.message.message_id, r.message.data.decode())
sub.acknowledge(subscription=pull_sub, ack_ids=[r.ack_id])
time.sleep(3) # Pub/Sub calls us
for body, code in got:
data = base64.b64decode(body["message"]["data"]).decode()
print("push:", body["message"]["messageId"], data,
"from", body["subscription"].split("/")[-1], "->", code)
sub.delete_subscription(subscription=push_sub)
""")
run('python pull_and_push.py')
# Step 2: Streaming pull with FlowControl(max_messages=3), then with the default limit
write('stream.py', """\
import sys, threading, time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import FlowControl
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "analytics-pull")
lock, s = threading.Lock(), {"now": 0, "peak": 0, "done": 0}
def callback(message):
with lock:
s["now"] += 1
s["peak"] = max(s["peak"], s["now"])
time.sleep(0.2) # slow work: 200 ms
message.ack()
with lock:
s["now"] -= 1
s["done"] += 1
flow = FlowControl(max_messages=int(sys.argv[1])) # the brake
future = sub.subscribe(path, callback=callback, flow_control=flow)
try:
future.result(timeout=6) # stream for 6 seconds
except TimeoutError:
future.cancel()
future.result()
print(f"max_messages={flow.max_messages}: processed {s['done']},",
f"peak in flight {s['peak']}")
""")
run('python publish.py clicks 20 2001')
run('python stream.py 3')
run('python publish.py clicks 20 3001')
run('python stream.py 1000')
# Step 3: Pull without ack: the message comes back after the 10 s ack deadline
write('deadline.py', """\
import time
from google.cloud import pubsub_v1
sub = pubsub_v1.SubscriberClient()
path = sub.subscription_path("tb-shop-dev", "invoices-pull")
config = sub.get_subscription(subscription=path)
print("ack deadline:", config.ack_deadline_seconds, "seconds")
resp = sub.pull(subscription=path, max_messages=1)
first = resp.received_messages[0]
print("pulled message", first.message.message_id, "- no ack")
start = time.time()
time.sleep(2)
early = sub.pull(subscription=path, max_messages=1, timeout=6)
n = len(early.received_messages)
print("pull before the deadline:", n, "messages")
time.sleep(max(0, 13 - (time.time() - start)))
resp = sub.pull(subscription=path, max_messages=1)
again = resp.received_messages[0]
print("pull after the deadline: message", again.message.message_id)
same = again.message.message_id == first.message.message_id
print("same message id:", same)
print("same ack id:", again.ack_id == first.ack_id)
sub.acknowledge(subscription=path, ack_ids=[again.ack_id])
""")
run('python publish.py invoices 1 5001')
run('python deadline.py')
# Step 4: A push endpoint that fails four times, then succeeds
write('push_retry.py', """\
import time
from google.cloud import pubsub_v1
import endpoint
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "refunds")
path = sub.subscription_path("tb-shop-dev", "refunds-push")
server, url, got = endpoint.start(codes=[500, 503, 429, 400, 204])
sub.create_subscription(name=path, topic=topic,
push_config={"push_endpoint": url})
pub.publish(topic, b'{"refund_id": 77}').result()
time.sleep(10)
for n, (body, code) in enumerate(got, 1):
message_id = body["message"]["messageId"]
print("attempt", n, "message", message_id, "->", code)
sub.delete_subscription(subscription=path)
""")
run('python push_retry.py')
# Step 5: Three workers share one subscription; a second subscription gets its own copy
write('workers.py', """\
import collections, threading, time
from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.types import FlowControl
path = "projects/tb-shop-dev/subscriptions/ledger-pull"
lock, seen = threading.Lock(), collections.Counter()
def callback(message):
time.sleep(0.1) # the work
with lock:
seen[message.data.decode()] += 1
message.ack()
workers = []
for _ in range(3): # three subscriber clients
client = pubsub_v1.SubscriberClient()
workers.append(client.subscribe(path, callback=callback,
flow_control=FlowControl(max_messages=2)))
time.sleep(8)
for w in workers:
w.cancel()
print("3 workers on ledger-pull: processed", sum(seen.values()),
"messages,", len(seen), "unique orders")
print("most deliveries of one order:", max(seen.values()))
""")
run('python publish.py payments 30 6001')
run('python workers.py')
run('python drain.py audit-pull')
# Step 6: One topic: a push subscription for Cloud Run and a pull subscription for a batch job
write('scenario.py', """\
import time
from google.cloud import pubsub_v1
import endpoint
pub, sub = pubsub_v1.PublisherClient(), pubsub_v1.SubscriberClient()
topic = pub.topic_path("tb-shop-dev", "orders")
fast = sub.subscription_path("tb-shop-dev", "shipping-run")
slow = sub.subscription_path("tb-shop-dev", "finance-batch")
server, url, got = endpoint.start() # stands in for Cloud Run
sub.create_subscription(name=fast, topic=topic,
push_config={"push_endpoint": url},
ack_deadline_seconds=60)
sub.create_subscription(name=slow, topic=topic)
for order_id in range(7001, 7006):
data = f'{{"order_id": {order_id}}}'.encode()
pub.publish(topic, data).result()
time.sleep(3)
print("shipping-run (push):", len(got), "orders reached the service")
print("finance-batch (pull): waiting for the batch job")
sub.delete_subscription(subscription=fast)
""")
run('python scenario.py')
# Step 7: Later, the batch job runs and drains its own backlog
time.sleep(2) # the video pauses here too
run('python drain.py finance-batch')
Part 4: Acknowledgements and Redelivery
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.